已有数据同步功能完成

.gitmodules

@@ -1,3 +1,7 @@
 [submodule "NtyCo"]
 	path = NtyCo
 	url = https://github.com/wangbojing/NtyCo.git
+[submodule "libbpf-bootstrap"]
+	path = libbpf-bootstrap
+	url = https://github.com/libbpf/libbpf-bootstrap
+	ignore = dirty

README.md

@@ -26,8 +26,11 @@ sudo apt install -y bpftrace
 
 git clone git@gitlab.0voice.com:lianyiheng/9.1-kvstore.git
 cd 9.1-kvstore/
+git submodule update --init --recursive
+./init.sh
 make
 
+
 ```
 
 ## 测试

config/config.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <config>
     <server>
-        <ip>127.0.0.1</ip>
+        <ip>192.168.10.129</ip>
         <port>8888</port>
         <mode>master</mode>     <!-- master / slave -->
 

diskuring/diskuring.c

@@ -3,8 +3,6 @@
 #include <poll.h>
 #include <sys/eventfd.h>
 
-iouring_ctx_t global_uring_ctx;
-
 void task_init(task_t *t)
 {
     pthread_mutex_init(&t->m, NULL);

dump/kvs_dump.h

@@ -2,15 +2,16 @@
 #define __KVS_DUMP_H__
 
 #include "kvstore.h"
+#include "diskuring/diskuring.h"
 
 extern char global_oplog_file[256];
 extern char global_array_file[256];
 extern char global_rbtree_file[256];
 extern char global_hash_file[256];
 
-int kvs_create_snapshot(const char* array_file, const char* rbtree_file, const char* hash_file);
+int kvs_create_snapshot(iouring_ctx_t *uring, const char* array_file, const char* rbtree_file, const char* hash_file);
 int kvs_create_snapshot_async(const char *ip, int port);
-void __create_snapshot_ok(const char* array_file, const char* rbtree_file, const char* hash_file);
+void __complete_snapshot(const char *ip, int port, const char *array_file, const char *rbtree_file, const char *hash_file);
 
 extern int global_oplog_fd;
 

dump/kvs_snapshot.c

@@ -1,16 +1,20 @@
 #include "kvstore.h"
+#include "diskuring/diskuring.h"
 #include <unistd.h>
+#include <arpa/inet.h>
+#include <sys/socket.h>
+#include <sys/types.h>
 
 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_create_snapshot(const char* array_file, const char* rbtree_file, const char* hash_file){
+int kvs_create_snapshot(iouring_ctx_t *uring, const char* array_file, const char* rbtree_file, const char* hash_file){
 	int ret = 0;
 	int rc = 0;
 	#if ENABLE_ARRAY
-	rc = kvs_array_save(&global_array, array_file);
+	rc = kvs_array_save(uring, &global_array, array_file);
 	if(rc < 0){
 		printf("kvs_engine_array save error\n");
 		ret = -1;
@@ -18,7 +22,7 @@ int kvs_create_snapshot(const char* array_file, const char* rbtree_file, const c
 	#endif
 
 	#if ENABLE_RBTREE
-	rc = kvs_rbtree_save(&global_rbtree, rbtree_file);
+	rc = kvs_rbtree_save(uring, &global_rbtree, rbtree_file);
 	if(rc < 0){
 		printf("kvs_engine_rbtree save error\n");
 		ret = -1;
@@ -26,7 +30,7 @@ int kvs_create_snapshot(const char* array_file, const char* rbtree_file, const c
 	#endif
 
 	#if ENABLE_HASH
-	rc = kvs_hash_save(&global_hash, hash_file);
+	rc = kvs_hash_save(uring, &global_hash, hash_file);
 	if(rc < 0){
 		printf("kvs_engine_hash save error\n");
 		ret = -1;
@@ -35,44 +39,93 @@ int kvs_create_snapshot(const char* array_file, const char* rbtree_file, const c
 	return ret;
 }
 
-void __create_snapshot_ok(const char* array_file, const char* rbtree_file, const char* hash_file){
+void __complete_snapshot(const char *ip, int port, const char *array_file, const char *rbtree_file, const char *hash_file){
 	
 }
 
-int kvs_create_snapshot_async(const char *ip, int port){
+static int send_file_to_ipport(const char *ip, int port, const char *filename) {
-	int pipefd[2];  // 用于子进程通知主进程
+	int sockfd = socket(AF_INET, SOCK_STREAM, 0);
-	if (pipe(pipefd) == -1) { perror("pipe"); return -1; }
+	if (sockfd < 0) { perror("socket"); return -1; }
-	pid_t pid = fork();
-    if (pid == -1) { perror("fork"); return -1; }
 
-	if (pid == 0) {  // 子进程
+	struct sockaddr_in addr;
-        close(pipefd[0]);  // 关闭读端
+    addr.sin_family = AF_INET;
-        // 指定临时文件路径，避免覆盖 global_xxx_file
+    addr.sin_port = htons(port);
-		char tmp_array[128];   // 可写缓冲区
+	addr.sin_addr.s_addr = inet_addr(ip);
-		char tmp_rbtree[128];
-		char tmp_hash[128];
 
-        snprintf(tmp_array,  sizeof(tmp_array),  "snapshot_array_%s.tmp", ip);
+	if (connect(sockfd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
-		snprintf(tmp_rbtree, sizeof(tmp_rbtree), "snapshot_rbtree_%s.tmp", ip);
+        perror("connect"); close(sockfd); return -1;
-		snprintf(tmp_hash,   sizeof(tmp_hash),   "snapshot_hash_%s.tmp", ip);
-
-        int ret = kvs_create_snapshot(tmp_array, tmp_rbtree, tmp_hash);
-        if (ret == 0) {
-            // 成功：rename 到最终路径，或直接通知
-            write(pipefd[1], "OK", 2);  // 通知主进程
-        } else {
-            write(pipefd[1], "ERR", 3);
-        }
-        close(pipefd[1]);
-        _exit(0);  // 子进程退出
-
-		// hook point 
-		__create_snapshot_ok(tmp_array, tmp_rbtree, tmp_hash);
-    } else {  // 主进程
-        close(pipefd[1]);  // 关闭写端
-        // 立即返回，继续处理其他请求（不阻塞）
-        // 可以记录 pid，在别处 waitpid(pid, NULL, WNOHANG) 检查完成
-        // 或用信号：signal(SIGCHLD, handler); 在 handler 中 read(pipefd[0]) 检查 "OK"
-        return 0;  // SYNC 响应成功，主进程继续
     }
+
+	FILE *fp = fopen(filename, "rb");
+    if (!fp) { perror("fopen"); close(sockfd); return -1; }
+
+	char buf[4096];
+	size_t n;
+	
+	while ((n = fread(buf, 1, sizeof(buf), fp)) > 0) {
+        ssize_t written = 0;
+        while (written < n) {
+            ssize_t ret = write(sockfd, buf + written, n - written);
+            if (ret < 0) { 
+                perror("write"); 
+                fclose(fp); 
+                close(sockfd); 
+                return -1; 
+            }
+            written += ret;
+        }
+    }
+
+	if (ferror(fp)) {
+        perror("fread");
+        fclose(fp);
+        close(sockfd);
+        return -1;
+    }
+
+	fclose(fp);
+	shutdown(sockfd, SHUT_WR);
+    close(sockfd);
+    return 0;
+}
+
+int kvs_create_snapshot_async(const char *ip, int port){
+	pid_t pid = fork();
+	if (pid == -1) { perror("fork"); return -1; }
+
+	if (pid == 0) {
+		char tmp_array[128];
+        char tmp_rbtree[128];
+        char tmp_hash[128];
+
+		snprintf(tmp_array,  sizeof(tmp_array),  "data/snapshot_array_%s.tmp.db", ip);
+        snprintf(tmp_rbtree, sizeof(tmp_rbtree), "data/snapshot_rbtree_%s.tmp.db", ip);
+        snprintf(tmp_hash,   sizeof(tmp_hash),   "data/snapshot_hash_%s.tmp.db", ip);
+
+		iouring_ctx_t uring;
+		iouring_init(&uring, 256);
+
+		int ret = kvs_create_snapshot(&uring, tmp_array, tmp_rbtree, tmp_hash);
+		if (ret != 0) {
+            fprintf(stderr, "snapshot creation failed\n");
+            _exit(1);
+        }
+
+		// 发送快照到目标 IP:port
+		if (send_file_to_ipport(ip, port, tmp_array) != 0 ||
+            send_file_to_ipport(ip, port, tmp_rbtree) != 0 ||
+            send_file_to_ipport(ip, port, tmp_hash) != 0) {
+            fprintf(stderr, "snapshot send failed\n");
+            _exit(1);
+        }
+
+		// hook 通知 eBPF
+		__complete_snapshot(ip, port, tmp_array, tmp_rbtree, tmp_hash);
+
+		iouring_shutdown(&uring);
+		_exit(0);
+	} else {
+
+		return 0;
+	}
 }

ebpf/c/.gitignore

@@ -0,0 +1,18 @@
+/.output
+/bootstrap
+/bootstrap_legacy
+/minimal
+/minimal_legacy
+/minimal_ns
+/uprobe
+/kprobe
+/fentry
+/profile
+/usdt
+/sockfilter
+/tc
+/ksyscall
+/task_iter
+/lsm
+/cmake-build-debug/
+/cmake-build-release/

ebpf/c/CMakeLists.txt

@@ -0,0 +1,133 @@
+# SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
+
+cmake_minimum_required(VERSION 3.16)
+project(examples C)
+
+# Tell cmake where to find BpfObject module
+list(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../../tools/cmake)
+
+# Build vendored libbpf
+include(ExternalProject)
+ExternalProject_Add(libbpf
+  PREFIX libbpf
+  SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/../../libbpf/src
+  CONFIGURE_COMMAND ""
+  BUILD_COMMAND make
+    CC=${CMAKE_C_COMPILER}
+    BUILD_STATIC_ONLY=1
+    OBJDIR=${CMAKE_CURRENT_BINARY_DIR}/libbpf/libbpf
+    DESTDIR=${CMAKE_CURRENT_BINARY_DIR}/libbpf
+    INCLUDEDIR=
+    LIBDIR=
+    UAPIDIR=
+    install install_uapi_headers
+  BUILD_IN_SOURCE TRUE
+  INSTALL_COMMAND ""
+  STEP_TARGETS build
+  BUILD_BYPRODUCTS ${CMAKE_CURRENT_BINARY_DIR}/libbpf/libbpf.a
+)
+
+ExternalProject_Add(bpftool
+  PREFIX bpftool
+  SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/../../bpftool/src
+  CONFIGURE_COMMAND ""
+  BUILD_COMMAND make bootstrap
+    OUTPUT=${CMAKE_CURRENT_BINARY_DIR}/bpftool/
+  BUILD_IN_SOURCE TRUE
+  INSTALL_COMMAND ""
+  STEP_TARGETS build
+)
+
+find_program(CARGO_EXISTS cargo)
+if(CARGO_EXISTS)
+  if(CMAKE_CROSSCOMPILING)
+    # Determine target triple
+    if(CMAKE_SYSTEM_NAME MATCHES "Linux")
+      if(CMAKE_SYSTEM_PROCESSOR MATCHES "x86_64")
+          set(CARGO_TARGET "x86_64-unknown-linux-gnu")
+      elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64")
+          set(CARGO_TARGET "aarch64-unknown-linux-gnu")
+      else()
+          message(FATAL_ERROR "Unsupported processor for Linux: ${CMAKE_SYSTEM_PROCESSOR}")
+      endif()
+
+      if(CMAKE_CXX_COMPILER)
+        set(RUST_LINKER ${CMAKE_CXX_COMPILER})
+      else()
+        set(RUST_LINKER ${CMAKE_C_COMPILER})
+      endif()
+    else()
+      message((FATAL_ERROR "Unsupported platform: ${CMAKE_SYSTEM_NAME}"))
+    endif()
+
+    ExternalProject_Add(blazesym
+      PREFIX blazesym
+      SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/../../blazesym
+      CONFIGURE_COMMAND ""
+      BUILD_COMMAND ${CMAKE_COMMAND} -E env
+        RUSTFLAGS=-C\ linker=${RUST_LINKER}
+        cargo build --package=blazesym-c --release --target=${CARGO_TARGET}
+      BUILD_IN_SOURCE TRUE
+      INSTALL_COMMAND ""
+      STEP_TARGETS build
+    )
+  else() # Host
+    ExternalProject_Add(blazesym
+      PREFIX blazesym
+      SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/../../blazesym
+      CONFIGURE_COMMAND ""
+      BUILD_COMMAND
+        cargo build --package=blazesym-c --release
+      BUILD_IN_SOURCE TRUE
+      INSTALL_COMMAND ""
+      STEP_TARGETS build
+    )
+  endif()
+endif()
+
+# Set BpfObject input parameters -- note this is usually not necessary unless
+# you're in a highly vendored environment (like libbpf-bootstrap)
+if(${CMAKE_SYSTEM_PROCESSOR} MATCHES "x86_64")
+  set(ARCH "x86")
+elseif(${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm")
+  set(ARCH "arm")
+elseif(${CMAKE_SYSTEM_PROCESSOR} MATCHES "aarch64")
+  set(ARCH "arm64")
+elseif(${CMAKE_SYSTEM_PROCESSOR} MATCHES "ppc64le")
+  set(ARCH "powerpc")
+elseif(${CMAKE_SYSTEM_PROCESSOR} MATCHES "mips")
+  set(ARCH "mips")
+elseif(${CMAKE_SYSTEM_PROCESSOR} MATCHES "riscv64")
+  set(ARCH "riscv")
+elseif(${CMAKE_SYSTEM_PROCESSOR} MATCHES "loongarch64")
+  set(ARCH "loongarch")
+endif()
+
+set(BPFOBJECT_BPFTOOL_EXE ${CMAKE_CURRENT_BINARY_DIR}/bpftool/bootstrap/bpftool)
+set(BPFOBJECT_VMLINUX_H ${CMAKE_CURRENT_SOURCE_DIR}/../../vmlinux.h/include/${ARCH}/vmlinux.h)
+set(LIBBPF_INCLUDE_DIRS ${CMAKE_CURRENT_BINARY_DIR}/libbpf)
+set(LIBBPF_LIBRARIES ${CMAKE_CURRENT_BINARY_DIR}/libbpf/libbpf.a)
+find_package(BpfObject REQUIRED)
+
+# Create an executable for each application
+file(GLOB apps *.bpf.c)
+if(NOT CARGO_EXISTS)
+  list(REMOVE_ITEM apps ${CMAKE_CURRENT_SOURCE_DIR}/profile.bpf.c)
+endif()
+foreach(app ${apps})
+  get_filename_component(app_stem ${app} NAME_WE)
+
+  # Build object skeleton and depend skeleton on libbpf build
+  bpf_object(${app_stem} ${app_stem}.bpf.c)
+  add_dependencies(${app_stem}_skel libbpf bpftool)
+
+  add_executable(${app_stem} ${app_stem}.c)
+  target_link_libraries(${app_stem} ${app_stem}_skel)
+  if(${app_stem} STREQUAL profile)
+    target_include_directories(${app_stem} PRIVATE
+      ${CMAKE_CURRENT_SOURCE_DIR}/../../blazesym/capi/include)
+    target_link_libraries(${app_stem}
+      ${CMAKE_CURRENT_SOURCE_DIR}/../../blazesym/target/${CARGO_TARGET}/release/libblazesym_c.a -lpthread -lrt -ldl)
+    add_dependencies(${app_stem} blazesym)
+  endif()
+endforeach()

ebpf/c/Makefile

@@ -0,0 +1,139 @@
+# SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause)
+OUTPUT := .output
+CLANG ?= clang
+LIBBPF_SRC := $(abspath ../../libbpf/src)
+BPFTOOL_SRC := $(abspath ../../bpftool/src)
+LIBBPF_OBJ := $(abspath $(OUTPUT)/libbpf.a)
+BPFTOOL_OUTPUT ?= $(abspath $(OUTPUT)/bpftool)
+BPFTOOL ?= $(BPFTOOL_OUTPUT)/bootstrap/bpftool
+LIBBLAZESYM_SRC := $(abspath ../../blazesym/)
+LIBBLAZESYM_INC := $(abspath $(LIBBLAZESYM_SRC)/capi/include)
+LIBBLAZESYM_OBJ := $(abspath $(OUTPUT)/libblazesym_c.a)
+ARCH ?= $(shell uname -m | sed 's/x86_64/x86/' \
+			 | sed 's/arm.*/arm/' \
+			 | sed 's/aarch64/arm64/' \
+			 | sed 's/ppc64le/powerpc/' \
+			 | sed 's/mips.*/mips/' \
+			 | sed 's/riscv64/riscv/' \
+			 | sed 's/loongarch64/loongarch/')
+VMLINUX := ../../vmlinux.h/include/$(ARCH)/vmlinux.h
+# Use our own libbpf API headers and Linux UAPI headers distributed with
+# libbpf to avoid dependency on system-wide headers, which could be missing or
+# outdated
+INCLUDES := -I$(OUTPUT) -I../../libbpf/include/uapi -I$(dir $(VMLINUX)) -I$(LIBBLAZESYM_INC)
+CFLAGS := -g -Wall
+ALL_LDFLAGS := $(LDFLAGS) $(EXTRA_LDFLAGS)
+
+# APPS = minimal minimal_legacy minimal_ns bootstrap bootstrap_legacy uprobe kprobe fentry \
+       usdt sockfilter tc ksyscall task_iter lsm 
+APPS = replica
+
+CARGO ?= $(shell which cargo)
+ifeq ($(strip $(CARGO)),)
+BZS_APPS :=
+else
+BZS_APPS := profile
+APPS += $(BZS_APPS)
+# Required by libblazesym
+ALL_LDFLAGS += -lrt -ldl -lpthread -lm
+endif
+
+# Get Clang's default includes on this system. We'll explicitly add these dirs
+# to the includes list when compiling with `-target bpf` because otherwise some
+# architecture-specific dirs will be "missing" on some architectures/distros -
+# headers such as asm/types.h, asm/byteorder.h, asm/socket.h, asm/sockios.h,
+# sys/cdefs.h etc. might be missing.
+#
+# Use '-idirafter': Don't interfere with include mechanics except where the
+# build would have failed anyways.
+CLANG_BPF_SYS_INCLUDES ?= $(shell $(CLANG) -v -E - </dev/null 2>&1 \
+	| sed -n '/<...> search starts here:/,/End of search list./{ s| \(/.*\)|-idirafter \1|p }')
+
+ifeq ($(V),1)
+	Q =
+	msg =
+else
+	Q = @
+	msg = @printf '  %-8s %s%s\n'					\
+		      "$(1)"						\
+		      "$(patsubst $(abspath $(OUTPUT))/%,%,$(2))"	\
+		      "$(if $(3), $(3))";
+	MAKEFLAGS += --no-print-directory
+endif
+
+define allow-override
+  $(if $(or $(findstring environment,$(origin $(1))),\
+            $(findstring command line,$(origin $(1)))),,\
+    $(eval $(1) = $(2)))
+endef
+
+$(call allow-override,CC,$(CROSS_COMPILE)cc)
+$(call allow-override,LD,$(CROSS_COMPILE)ld)
+
+.PHONY: all
+all: $(APPS)
+
+.PHONY: clean
+clean:
+	$(call msg,CLEAN)
+	$(Q)rm -rf $(OUTPUT) $(APPS)
+
+$(OUTPUT) $(OUTPUT)/libbpf $(BPFTOOL_OUTPUT):
+	$(call msg,MKDIR,$@)
+	$(Q)mkdir -p $@
+
+# Build libbpf
+$(LIBBPF_OBJ): $(wildcard $(LIBBPF_SRC)/*.[ch] $(LIBBPF_SRC)/Makefile) | $(OUTPUT)/libbpf
+	$(call msg,LIB,$@)
+	$(Q)$(MAKE) -C $(LIBBPF_SRC) BUILD_STATIC_ONLY=1		      \
+		    OBJDIR=$(dir $@)/libbpf DESTDIR=$(dir $@)		      \
+		    INCLUDEDIR= LIBDIR= UAPIDIR=			      \
+		    install
+
+# Build bpftool
+$(BPFTOOL): | $(BPFTOOL_OUTPUT)
+	$(call msg,BPFTOOL,$@)
+	$(Q)$(MAKE) ARCH= CROSS_COMPILE= OUTPUT=$(BPFTOOL_OUTPUT)/ -C $(BPFTOOL_SRC) bootstrap
+
+
+$(LIBBLAZESYM_SRC)/target/release/libblazesym_c.a::
+	$(Q)cd $(LIBBLAZESYM_SRC) && $(CARGO) build --package=blazesym-c --release
+
+$(LIBBLAZESYM_OBJ): $(LIBBLAZESYM_SRC)/target/release/libblazesym_c.a | $(OUTPUT)
+	$(call msg,LIB, $@)
+	$(Q)cp $(LIBBLAZESYM_SRC)/target/release/libblazesym_c.a $@
+
+# Build BPF code
+$(OUTPUT)/%.bpf.o: %.bpf.c $(LIBBPF_OBJ) $(wildcard %.h) $(VMLINUX) | $(OUTPUT) $(BPFTOOL)
+	$(call msg,BPF,$@)
+	$(Q)$(CLANG) -g -O2 -target bpf -D__TARGET_ARCH_$(ARCH)		      \
+		     $(INCLUDES) $(CLANG_BPF_SYS_INCLUDES)		      \
+		     -c $(filter %.c,$^) -o $(patsubst %.bpf.o,%.tmp.bpf.o,$@)
+	$(Q)$(BPFTOOL) gen object $@ $(patsubst %.bpf.o,%.tmp.bpf.o,$@)
+
+# Generate BPF skeletons
+$(OUTPUT)/%.skel.h: $(OUTPUT)/%.bpf.o | $(OUTPUT) $(BPFTOOL)
+	$(call msg,GEN-SKEL,$@)
+	$(Q)$(BPFTOOL) gen skeleton $< > $@
+
+# Build user-space code
+$(patsubst %,$(OUTPUT)/%.o,$(APPS)): %.o: %.skel.h
+
+$(OUTPUT)/%.o: %.c $(wildcard %.h) | $(OUTPUT)
+	$(call msg,CC,$@)
+	$(Q)$(CC) $(CFLAGS) $(INCLUDES) -c $(filter %.c,$^) -o $@
+
+$(patsubst %,$(OUTPUT)/%.o,$(BZS_APPS)): $(LIBBLAZESYM_OBJ)
+
+$(BZS_APPS): $(LIBBLAZESYM_OBJ)
+
+# Build application binary
+$(APPS): %: $(OUTPUT)/%.o $(LIBBPF_OBJ) | $(OUTPUT)
+	$(call msg,BINARY,$@)
+	$(Q)$(CC) $(CFLAGS) $^ $(ALL_LDFLAGS) -lelf -lz -o $@
+
+# delete failed targets
+.DELETE_ON_ERROR:
+
+# keep intermediate (.skel.h, .bpf.o, etc) targets
+.SECONDARY:

ebpf/c/replica.bpf.c

@@ -0,0 +1,103 @@
+// SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
+/* Copyright (c) 2020 Facebook */
+#include <linux/bpf.h>
+#include <linux/ptrace.h>
+#include <bpf/bpf_helpers.h>
+#include <bpf/bpf_tracing.h>
+
+#include "replica.h"
+
+char LICENSE[] SEC("license") = "Dual BSD/GPL";
+
+int my_pid = 0;
+
+struct {
+	__uint(type, BPF_MAP_TYPE_PERF_EVENT_ARRAY);
+	__uint(key_size, sizeof(int));
+	__uint(value_size, sizeof(int));
+} channel SEC(".maps");
+
+SEC("uprobe/kvs_create_snapshot_async")
+int uprobe_create_snapshot_async(struct pt_regs *ctx)
+{
+    struct event ev;
+	__builtin_memset(&ev, 0, sizeof(ev));
+
+    const char *ip;
+    __u32 port;
+
+    ev.type = EVENT_CREATE_SNAPSHOT_ASYNC;
+
+    ip   = (const char *)PT_REGS_PARM1(ctx);
+    port = (__u32)PT_REGS_PARM2(ctx);
+
+    bpf_probe_read_user_str(ev.data.sync.ip,
+                            sizeof(ev.data.sync.ip),
+                            ip);
+    ev.data.sync.port = port;
+
+    bpf_perf_event_output(ctx, &channel,
+                          BPF_F_CURRENT_CPU,
+                          &ev, sizeof(ev));
+    return 0;
+}
+
+SEC("uprobe/__compeleted_cmd")
+int uprobe_completed_cmd(struct pt_regs *ctx)
+{
+    struct event ev;
+	__builtin_memset(&ev, 0, sizeof(ev));
+
+    const __u8 *cmd;
+    __u32 len;
+
+    ev.type = EVENT_COMPLETED_CMD;
+
+    cmd = (const __u8 *)PT_REGS_PARM1(ctx);
+    len = (__u32)PT_REGS_PARM2(ctx);
+
+    if (len > sizeof(ev.data.cmd.cmd))
+        len = sizeof(ev.data.cmd.cmd);
+
+    ev.data.cmd.len = len;
+
+    bpf_probe_read_user(ev.data.cmd.cmd, len, cmd);
+
+    bpf_perf_event_output(ctx, &channel,
+                          BPF_F_CURRENT_CPU,
+                          &ev, sizeof(ev));
+    return 0;
+}
+
+
+SEC("uprobe/__create_snapshot_ok")
+int uprobe_create_snapshot_ok(struct pt_regs *ctx)
+{
+    struct event ev;
+	__builtin_memset(&ev, 0, sizeof(ev));
+	
+    const char *array_file;
+    const char *rbtree_file;
+    const char *hash_file;
+
+    ev.type = EVENT_CREATE_SNAPSHOT_OK;
+
+    array_file  = (const char *)PT_REGS_PARM1(ctx);
+    rbtree_file = (const char *)PT_REGS_PARM2(ctx);
+    hash_file   = (const char *)PT_REGS_PARM3(ctx);
+
+    bpf_probe_read_user_str(ev.data.ok.array_file,
+                            sizeof(ev.data.ok.array_file),
+                            array_file);
+    bpf_probe_read_user_str(ev.data.ok.rbtree_file,
+                            sizeof(ev.data.ok.rbtree_file),
+                            rbtree_file);
+    bpf_probe_read_user_str(ev.data.ok.hash_file,
+                            sizeof(ev.data.ok.hash_file),
+                            hash_file);
+
+    bpf_perf_event_output(ctx, &channel,
+                          BPF_F_CURRENT_CPU,
+                          &ev, sizeof(ev));
+    return 0;
+}

ebpf/c/replica.c

@@ -0,0 +1,257 @@
+// SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause)
+/* Copyright (c) 2020 Facebook */
+#include <stdio.h>
+#include <unistd.h>
+#include <sys/resource.h>
+#include <bpf/libbpf.h>
+#include "replica.skel.h"
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <fcntl.h>
+
+#include "replica.h"
+
+
+struct cmd_node {
+    uint8_t *cmd;
+    size_t len;
+    struct cmd_node *next;
+};
+
+struct pending_queue {
+    struct cmd_node *head;
+    struct cmd_node *tail;
+	int count;
+};
+
+static void queue_init(struct pending_queue *q) {
+    q->head = q->tail = NULL;
+	q->count = 0;
+}
+
+static void queue_push(struct pending_queue *q, const uint8_t *cmd, size_t len) {
+    struct cmd_node *node = malloc(sizeof(*node));
+    if (!node) return;
+    node->cmd = malloc(len);
+    if (!node->cmd) { free(node); return; }
+    memcpy(node->cmd, cmd, len);
+    node->len = len;
+    node->next = NULL;
+    if (q->tail) q->tail->next = node;
+    else q->head = node;
+    q->tail = node;
+    q->count++;
+}
+
+static void queue_send_and_clear(struct pending_queue *q, int sock) {
+    struct cmd_node *node = q->head;
+    int sent_count = 0;
+    while (node) {
+        if (send(sock, node->cmd, node->len, 0) > 0) {
+            sent_count++;
+        }
+        struct cmd_node *tmp = node;
+        node = node->next;
+        free(tmp->cmd);
+        free(tmp);
+    }
+    if (sent_count > 0) {
+        printf("[QUEUE] Sent %d commands to slave\n", sent_count);
+    }
+    queue_init(q);
+}
+
+static void queue_free(struct pending_queue *q) {
+    struct cmd_node *node = q->head;
+    while (node) {
+        struct cmd_node *tmp = node;
+        node = node->next;
+        free(tmp->cmd);
+        free(tmp);
+    }
+    queue_init(q);
+}
+
+static int send_file(int sock, const char *path) {
+    FILE *fp = fopen(path, "rb");
+    if (!fp) {
+        printf("[ERROR] Failed to open file: %s\n", path);
+        return -1;
+    }
+    
+    char buf[4096];
+    size_t n, total = 0;
+    while ((n = fread(buf, 1, sizeof(buf), fp)) > 0) {
+        if (send(sock, buf, n, 0) < 0) { 
+            fclose(fp); 
+            printf("[ERROR] Failed to send file: %s (sent %zu bytes)\n", path, total);
+            return -1; 
+        }
+        total += n;
+    }
+    fclose(fp);
+    printf("[FILE] Sent %s (%zu bytes)\n", path, total);
+    return 0;
+}
+
+// 全局状态（单 Slave 简化）
+static enum state current_state = NOSLAVE;
+static char slave_ip[16] = {0};
+static int slave_port = 0;
+static char array_file[128] = {0};
+static char rbtree_file[128] = {0};
+static char hash_file[128] = {0};
+static struct pending_queue pending;
+static int slave_sock = -1;  // 连接 Slave 的 socket
+
+// 连接 Slave
+static int connect_slave() {
+    int sock = socket(AF_INET, SOCK_STREAM, 0);
+    if (sock < 0) return -1;
+    struct sockaddr_in addr;
+    memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(slave_port);
+    inet_pton(AF_INET, slave_ip, &addr.sin_addr);
+    if (connect(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
+        close(sock);
+        return -1;
+    }
+    return sock;
+}
+
+static void handle_event(void *ctx, int cpu, void *data, __u32 size) {
+    struct event *ev = (struct event *)data;
+
+    switch (ev->type) {
+        case EVENT_CREATE_SNAPSHOT_ASYNC:
+			printf("[EVENT] Type: CREATE_SNAPSHOT_ASYNC\n");
+            printf("[EVENT] Slave IP: %s, Port: %u\n", ev->data.sync.ip, ev->data.sync.port);
+
+            if (current_state == NOSLAVE) {
+                current_state = START;
+                strncpy(slave_ip, ev->data.sync.ip, sizeof(slave_ip));
+                slave_port = ev->data.sync.port;
+                queue_init(&pending);
+                slave_sock = connect_slave();  // 连接 Slave
+                if (slave_sock < 0) {
+                    printf("Failed to connect to Slave %s:%d\n", slave_ip, slave_port);
+                    current_state = NOSLAVE;
+                }
+            }
+            break;
+        case EVENT_COMPLETED_CMD:
+			printf("[EVENT] Type: COMPLETED_CMD\n");
+            printf("[EVENT] Command length: %llu bytes\n", ev->data.cmd.len);
+
+            if (current_state != NOSLAVE) {
+                queue_push(&pending, ev->data.cmd.cmd, ev->data.cmd.len);
+            }
+            break;
+        case EVENT_CREATE_SNAPSHOT_OK:
+			printf("[EVENT] Type: CREATE_SNAPSHOT_OK\n");
+            printf("[EVENT] Array file: %s\n", ev->data.ok.array_file);
+            printf("[EVENT] RBTree file: %s\n", ev->data.ok.rbtree_file);
+            printf("[EVENT] Hash file: %s\n", ev->data.ok.hash_file);
+
+            if (current_state == START) {
+                current_state = DONE;
+                strncpy(array_file, ev->data.ok.array_file, sizeof(array_file));
+                strncpy(rbtree_file, ev->data.ok.rbtree_file, sizeof(rbtree_file));
+                strncpy(hash_file, ev->data.ok.hash_file, sizeof(hash_file));
+            }
+            break;
+    }
+}
+
+
+static void lost_event(void *ctx, int cpu, __u64 cnt) {
+    printf("Lost %llu events\n", cnt);
+}
+
+static int libbpf_print_fn(enum libbpf_print_level level, const char *format, va_list args)
+{
+	return vfprintf(stderr, format, args);
+}
+
+int main(int argc, char **argv)
+{
+	struct replica_bpf *skel;
+	int err;
+
+	/* Set up libbpf errors and debug info callback */
+	libbpf_set_print(libbpf_print_fn);
+
+	/* Open BPF application */
+	skel = replica_bpf__open();
+	if (!skel) {
+		fprintf(stderr, "Failed to open BPF skeleton\n");
+		return 1;
+	}
+
+	/* ensure BPF program only handles write() syscalls from our process */
+	skel->bss->my_pid = getpid();
+
+	/* Load & verify BPF programs */
+	err = replica_bpf__load(skel);
+	if (err) {
+		fprintf(stderr, "Failed to load and verify BPF skeleton\n");
+		goto cleanup;
+	}
+
+	/* Attach tracepoint handler */
+	err = replica_bpf__attach(skel);
+	if (err) {
+		fprintf(stderr, "Failed to attach BPF skeleton\n");
+		goto cleanup;
+	}
+
+	printf("Successfully started! Please run `sudo cat /sys/kernel/debug/tracing/trace_pipe` "
+	       "to see output of the BPF programs.\n");
+
+
+	struct perf_buffer *pb = perf_buffer__new(bpf_map__fd(skel->maps.channel), 8, handle_event, lost_event, NULL, NULL);
+	if(!pb){
+		goto cleanup;
+	}
+
+#if 0
+	while(1){
+		perf_buffer__poll(pb, 1000);
+	}
+#else 
+	while (1) {
+        perf_buffer__poll(pb, 1000);  // 处理事件
+
+        // 循环中检查状态并发送
+        if (current_state == DONE && slave_sock >= 0) {
+            // 发送快照文件
+            if (send_file(slave_sock, array_file) == 0 &&
+                send_file(slave_sock, rbtree_file) == 0 &&
+                send_file(slave_sock, hash_file) == 0) {
+                current_state = ONLINE;
+                printf("Snapshot sent, state to ONLINE\n");
+            } else {
+                printf("Failed to send snapshot\n");
+                current_state = NOSLAVE;
+                close(slave_sock);
+                slave_sock = -1;
+            }
+        }
+
+        if (current_state == ONLINE && slave_sock >= 0) {
+            // 发送 pending
+            queue_send_and_clear(&pending, slave_sock);
+        }
+    }
+#endif
+
+	perf_buffer__free(pb);
+
+cleanup:
+	queue_free(&pending);
+	if (slave_sock >= 0) close(slave_sock);
+	replica_bpf__destroy(skel);
+	return -err;
+}

ebpf/c/replica.h

@@ -0,0 +1,36 @@
+#ifndef __REPLICA_H__
+#define __REPLICA_H__
+
+enum event_type {
+    EVENT_CREATE_SNAPSHOT_ASYNC,
+    EVENT_CREATE_SNAPSHOT_OK,
+    EVENT_CREATE_SNAPSHOT_READY,
+    EVENT_COMPLETED_CMD
+};
+
+struct event {
+    enum event_type type;
+    union {
+        struct {
+            char ip[16];
+            __u32 port;
+        } sync;
+        struct {
+            __u8 cmd[256];
+            __u64 len;
+        } cmd;
+        struct {
+            char array_file[128];
+            char rbtree_file[128];
+            char hash_file[128];
+        } ok;
+    } data;
+};
+
+enum state {
+    NOSLAVE,
+    PREPARING,
+    ONLINE
+};
+
+#endif

ebpf/c/xmake.lua

@@ -0,0 +1,126 @@
+add_rules("mode.release", "mode.debug")
+add_rules("platform.linux.bpf")
+set_license("GPL-2.0")
+
+if xmake.version():satisfies(">=2.5.7 <=2.5.9") then
+    on_load(function (target)
+        raise("xmake(%s) has a bug preventing BPF source code compilation. Please run `xmake update -f 2.5.6` to revert to v2.5.6 version or upgrade to xmake v2.6.1 that fixed the issue.", xmake.version())
+    end)
+end
+
+option("system-libbpf",      {showmenu = true, default = false, description = "Use system-installed libbpf"})
+option("require-bpftool",    {showmenu = true, default = false, description = "Require bpftool package"})
+
+add_requires("elfutils", "zlib")
+if is_plat("android") then
+    add_requires("ndk >=22.x <26", "argp-standalone")
+    set_toolchains("@ndk", {sdkver = "23"})
+else
+    add_requires("llvm >=10.x")
+    set_toolchains("@llvm")
+    add_requires("linux-headers")
+end
+
+-- fix error: libbpf: map 'my_pid_map': unsupported map linkage static. for bpftool >= 7.2.0
+-- we cannot add `"-fvisibility=hidden"` when compiling *.bpf.c
+set_symbols("none")
+
+if is_arch("arm64", "arm64-v8a") then
+    add_includedirs("../../vmlinux.h/include/arm64")
+elseif is_arch("arm.*") then
+    add_includedirs("../../vmlinux.h/include/arm")
+elseif is_arch("riscv32", "riscv64") then
+    add_includedirs("../../vmlinux.h/include/riscv")
+elseif is_arch("loongarch") then
+    add_includedirs("../../vmlinux.h/include/loongarch")
+elseif is_arch("ppc", "powerpc") then
+    add_includedirs("../../vmlinux.h/include/powerpc")
+elseif is_arch("x86_64", "i386") then
+    add_includedirs("../../vmlinux.h/include/x86")
+else
+    add_includedirs("../../vmlinux.h/include")
+end
+
+-- we can run `xmake f --require-bpftool=y` to pull bpftool from xmake-repo repository
+if has_config("require-bpftool") then
+    add_requires("linux-tools", {configs = {bpftool = true}})
+    add_packages("linux-tools")
+else
+    before_build(function (target)
+        os.addenv("PATH", path.join(os.scriptdir(), "..", "..", "tools"))
+    end)
+end
+
+-- we use the vendored libbpf sources for libbpf-bootstrap.
+-- for some projects you may want to use the system-installed libbpf, so you can run `xmake f --system-libbpf=y`
+if has_config("system-libbpf") then
+    add_requires("libbpf", {system = true})
+else
+    target("libbpf")
+        set_kind("static")
+        set_basename("bpf")
+        add_files("../../libbpf/src/*.c")
+        add_includedirs("../../libbpf/include")
+        add_includedirs("../../libbpf/include/uapi", {public = true})
+        add_includedirs("$(buildir)", {interface = true})
+        add_configfiles("../../libbpf/src/(*.h)", {prefixdir = "bpf"})
+        add_packages("elfutils", "zlib")
+        if is_plat("android") then
+            add_defines("__user=", "__force=", "__poll_t=uint32_t")
+        end
+end
+
+target("minimal")
+    set_kind("binary")
+    add_files("minimal.c", "minimal.bpf.c")
+    add_packages("linux-headers")
+    if not has_config("system-libbpf") then
+        add_deps("libbpf")
+    end
+
+target("minimal_legacy")
+    set_kind("binary")
+    add_files("minimal_legacy.c", "minimal_legacy.bpf.c")
+    add_packages("linux-headers")
+    if not has_config("system-libbpf") then
+        add_deps("libbpf")
+    end
+
+target("bootstrap")
+    set_kind("binary")
+    add_files("bootstrap.c", "bootstrap.bpf.c")
+    add_packages("linux-headers")
+    if not has_config("system-libbpf") then
+        add_deps("libbpf")
+    end
+    if is_plat("android") then
+        add_packages("argp-standalone")
+    end
+
+target("fentry")
+    set_kind("binary")
+    add_files("fentry.c", "fentry.bpf.c")
+    add_packages("linux-headers")
+    if not has_config("system-libbpf") then
+        add_deps("libbpf")
+    end
+
+target("uprobe")
+    set_kind("binary")
+    add_files("uprobe.c", "uprobe.bpf.c")
+    add_packages("linux-headers")
+    if not has_config("system-libbpf") then
+        add_deps("libbpf")
+    end
+
+target("kprobe")
+    set_kind("binary")
+    add_files("kprobe.c", "kprobe.bpf.c")
+    add_packages("linux-headers")
+    if not has_config("system-libbpf") then
+        add_deps("libbpf")
+    end
+    if is_plat("android") then
+        -- TODO we need fix vmlinux.h to support android
+        set_default(false)
+    end

init.sh

@@ -0,0 +1,7 @@
+rm -rf libbpf-bootstrap/examples/c
+
+cp -R ebpf/c libbpf-bootstrap/examples
+
+cd libbpf-bootstrap/examples/c
+
+make 

kvs_array_bin.c

@@ -191,8 +191,8 @@ int kvs_array_exist_bin(kvs_array_t *inst, const void *key, uint32_t key_len) {
 }
 
 // return: 0 success, <0 error
-int kvs_array_save(kvs_array_t *inst, const char* filename){
+int kvs_array_save(iouring_ctx_t *uring, kvs_array_t *inst, const char* filename){
-	if(!inst || !filename) return -1;
+	if(!uring || !inst || !filename) return -1;
 
 	int fd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, 0644);
     if(fd < 0) return -2;
@@ -234,7 +234,7 @@ int kvs_array_save(kvs_array_t *inst, const char* filename){
         size_t total = 0;
         for (int i = 0; i < count; i++) total += lens[i];
 
-        task_t *t = submit_write(&global_uring_ctx, fd, bufs, lens, count, current_off);
+        task_t *t = submit_write(uring, fd, bufs, lens, count, current_off);
 
 		if (!t) { close(fd); return -4; }
         int res = task_wait(t);

kvs_hash_bin.c

@@ -263,7 +263,7 @@ int kvs_hash_exist_bin(kvs_hash_t *hash, const void *key, uint32_t key_len) {
 }
 
 // 0 suc, <0 error
-int kvs_hash_save(kvs_hash_t *inst, const char* filename){
+int kvs_hash_save(iouring_ctx_t *uring, kvs_hash_t *inst, const char* filename){
 	if(!inst || !filename) return -1;
 	int fd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, 0644);
     if(fd < 0) return -2;
@@ -307,7 +307,7 @@ int kvs_hash_save(kvs_hash_t *inst, const char* filename){
 			for (int i = 0; i < count; i++) total += lens[i];
 
 
-			task_t *t = submit_write(&global_uring_ctx, fd, bufs, lens, count, current_off);
+			task_t *t = submit_write(uring, fd, bufs, lens, count, current_off);
 
 			if (!t) { close(fd); return -4; }
 			int res = task_wait(t);

kvs_protocol_resp.c

@@ -336,7 +336,7 @@ const char *command[] = {
 	"SET", "GET", "DEL", "MOD", "EXIST",
 	"RSET", "RGET", "RDEL", "RMOD", "REXIST",
 	"HSET", "HGET", "HDEL", "HMOD", "HEXIST",
-	"SAVE", "PSYNC"
+	"SAVE", "SSYNC", "SREADY"
 };
 
 
@@ -536,16 +536,19 @@ int resp_dispatch(const resp_cmd_t *cmd, resp_value_t *out_value) {
     /* ---------------- misc ---------------- */
     case KVS_CMD_SAVE: {
         if (cmd->argc != 1) { *out_value = resp_error("ERR wrong number of arguments for 'save'"); return 0; }
-        int r = kvs_create_snapshot(global_array_file, global_rbtree_file, global_hash_file);
+        int r = kvs_create_snapshot(&global_uring_ctx, global_array_file, global_rbtree_file, global_hash_file);
         if(r == 0) ksv_clear_log(global_oplog_fd);
         if (r < 0) { *out_value = resp_error("ERR save failed"); return 0; }
         *out_value = resp_simple("OK");
         return 0;
     }
-	case KVS_CMD_PSYNC:
+	case KVS_CMD_SSYNC:
         kvs_create_snapshot_async(cmd->argv[1].ptr, atoi(cmd->argv[2].ptr));
 		*out_value = resp_simple("OK");
 		return 0;
+    case KVS_CMD_SREADY:
+		*out_value = resp_simple("OK");
+		return 0;
 	default:
 		break;
 	}

kvs_protocol_resp.h

@@ -37,7 +37,8 @@ typedef enum {
 	KVS_CMD_HEXIST,
 	
 	KVS_CMD_SAVE,
-	KVS_CMD_PSYNC,
+	KVS_CMD_SSYNC,
+	KVS_CMD_SREADY,
 	KVS_CMD_COUNT,
 }kvs_cmd_t;
 

kvs_rbtree_bin.c

@@ -464,13 +464,13 @@ int kvs_rbtree_exist(kvs_rbtree_t *inst, const void *key, uint32_t key_len) {
 	return 0;
 }
 
-static int kvs_rbtree_save_node(int fd, off_t *current_off, kvs_rbtree_t *inst, rbtree_node *node) {
+static int kvs_rbtree_save_node(iouring_ctx_t *uring, int fd, off_t *current_off, kvs_rbtree_t *inst, rbtree_node *node) {
     if (!current_off || !inst || !node) return -1;
     if (node == inst->nil) return 0;
 
     int rc = 0;
 
-    rc = kvs_rbtree_save_node(fd, current_off, inst, node->left);
+    rc = kvs_rbtree_save_node(uring, fd, current_off, inst, node->left);
     if (rc < 0) return rc;
 
     uint32_t klen = htonl(node->key_len);
@@ -503,7 +503,7 @@ static int kvs_rbtree_save_node(int fd, off_t *current_off, kvs_rbtree_t *inst,
 	size_t total = 0;
     for (int i = 0; i < count; i++) total += lens[i];
 
-	task_t *t = submit_write(&global_uring_ctx, fd, bufs, lens, count, *current_off);
+	task_t *t = submit_write(uring, fd, bufs, lens, count, *current_off);
 
 	
 	if (!t) { return -4; }
@@ -516,22 +516,22 @@ static int kvs_rbtree_save_node(int fd, off_t *current_off, kvs_rbtree_t *inst,
 	
 	*current_off += (off_t) total;
 
-    rc = kvs_rbtree_save_node(fd, current_off, inst, node->right);
+    rc = kvs_rbtree_save_node(uring, fd, current_off, inst, node->right);
     if (rc < 0) return rc;
 
     return 0;
 }
 
 // 0 success, <0 error
-int kvs_rbtree_save(kvs_rbtree_t *inst, const char* filename){
+int kvs_rbtree_save(iouring_ctx_t *uring, kvs_rbtree_t *inst, const char* filename){
-	if (!inst || !filename) return -1;
+	if (!uring || !inst || !filename) return -1;
 
 	int fd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, 0644);
 	if(fd < 0) return -2;
     off_t current_off = 0;
 
 
-    int rc = kvs_rbtree_save_node(fd, &current_off, inst, inst->root);
+    int rc = kvs_rbtree_save_node(uring, fd, &current_off, inst, inst->root);
 
     close(fd);
     return rc; 

kvs_slave.c

@@ -3,40 +3,56 @@
 #include <string.h>
 #include <unistd.h>
 #include <arpa/inet.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <errno.h>
 
-/* 创建并监听用于接收快照的 TCP 监听 socket，成功返回 listen fd，失败返回 -1 */
-static int create_listen_socket(const char *ip, int port){
 
+static int recv_exact(int sockfd, char *buf, size_t len) {
+    size_t received = 0;
+    while (received < len) {
+        ssize_t ret = recv(sockfd, buf + received, len - received, 0);
+        if (ret <= 0) {
+            if (ret == 0) {
+                fprintf(stderr, "connection closed by peer\n");
+            } else {
+                perror("recv");
+            }
+            return -1;
+        }
+        received += ret;
+    }
+    return 0;
 }
 
-/* 主动连接 master 指定地址，用于控制面通信（SSYNC / SREADY），返回连接 fd 或 -1 */
+static int receive_snapshot_file(int sockfd, const char *filename) {
-static int connect_master(const char *master_ip, int master_port){
+    FILE *fp = fopen(filename, "wb");
-
+    if (!fp) {
-}
+        perror("fopen");
-
+        return -1;
-/* 通过控制连接向 master 发送 SSYNC 请求，声明本 slave 的快照接收地址 */
+    }
-static int send_ssync(int ctrl_fd, const char *listen_ip, int listen_port){
-
-}
-
-/* 接收并校验 master 对 SSYNC 的确认响应（如 +OK），成功返回 0 */
-static int recv_ssync_ok(int ctrl_fd){
-
-}
-
-/* 在快照监听 socket 上阻塞等待 master 的快照发送连接，返回已建立连接的 fd */
-static int accept_snapshot_conn(int listen_fd){
-
-}
-
-/* 从快照连接中接收完整快照数据并构建内存状态，确保快照已完全应用 */
-static int recv_and_apply_snapshot(int snapshot_fd){
-
-}
-
-/* 通过控制连接向 master 发送 SREADY 通知，表示快照已应用，slave 即将进入服务态 */
-int send_sready(int ctrl_fd){
 
+    char buf[4096];
+    ssize_t n;
+    
+    while ((n = recv(sockfd, buf, sizeof(buf), 0)) > 0) {
+        if (fwrite(buf, 1, n, fp) != n) {
+            perror("fwrite");
+            fclose(fp);
+            return -1;
+        }
+    }
+    
+    if (n < 0) {
+        perror("recv");
+        fclose(fp);
+        return -1;
+    }
+    
+    fclose(fp);
+    return 0;
 }
 
 
@@ -47,64 +63,155 @@ int slave_bootstrap(
     int master_port
 ) {
     int listen_fd = -1;
-    int ctrl_fd   = -1;
+    int master_fd = -1;
-    int snap_fd   = -1;
+    int snapshot_fds[3] = {-1, -1, -1};
+    int ret = -1;
 
-    /* 1. 监听 snapshot port */
+    // 1. 创建监听socket
-    listen_fd = create_listen_socket(listen_ip, listen_port);
+    listen_fd = socket(AF_INET, SOCK_STREAM, 0);
     if (listen_fd < 0) {
-        goto fail;
+        perror("socket");
+        goto cleanup;
     }
 
-    /* 2. 连接 master, 发送 SSYNC */
+    // 设置SO_REUSEADDR
-    ctrl_fd = connect_master(master_ip, master_port);
+    int opt = 1;
-    if (ctrl_fd < 0) {
+    if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
-        goto fail;
+        perror("setsockopt");
+        goto cleanup;
     }
 
-    if (send_ssync(ctrl_fd, listen_ip, listen_port) < 0) {
+    struct sockaddr_in listen_addr;
-        goto fail;
+    memset(&listen_addr, 0, sizeof(listen_addr));
+    listen_addr.sin_family = AF_INET;
+    listen_addr.sin_port = htons(listen_port);
+    listen_addr.sin_addr.s_addr = inet_addr(listen_ip);
+
+    if (bind(listen_fd, (struct sockaddr*)&listen_addr, sizeof(listen_addr)) < 0) {
+        perror("bind");
+        goto cleanup;
     }
 
-    if (recv_ssync_ok(ctrl_fd) < 0) {
+    if (listen(listen_fd, 5) < 0) {
-        goto fail;
+        perror("listen");
+        goto cleanup;
     }
 
-    close(ctrl_fd);
+    printf("Slave listening on %s:%d\n", listen_ip, listen_port);
-    ctrl_fd = -1;
 
-    /* 3. accept snapshot 连接 */
+
-    snap_fd = accept_snapshot_conn(listen_fd);
+    // 2. 连接master并发送SSYNC命令
-    if (snap_fd < 0) {
+    master_fd = socket(AF_INET, SOCK_STREAM, 0);
-        goto fail;
+    if (master_fd < 0) {
+        perror("socket");
+        goto cleanup;
     }
 
-    /* 4. 接收 snapshot */
+    struct sockaddr_in master_addr;
-    if (recv_and_apply_snapshot(snap_fd) < 0) {
+    memset(&master_addr, 0, sizeof(master_addr));
-        goto fail;
+    master_addr.sin_family = AF_INET;
+    master_addr.sin_port = htons(master_port);
+    master_addr.sin_addr.s_addr = inet_addr(master_ip);
+
+    if (connect(master_fd, (struct sockaddr*)&master_addr, sizeof(master_addr)) < 0) {
+        perror("connect to master");
+        goto cleanup;
     }
 
-    close(snap_fd);
+    printf("Connected to master %s:%d\n", master_ip, master_port);
-    snap_fd = -1;
+
+    // 构造RESP协议消息: SSYNC listen_ip listen_port
+    char ssync_cmd[256];
+    int cmd_len = snprintf(ssync_cmd, sizeof(ssync_cmd),
+                          "*3\r\n$5\r\nSSYNC\r\n$%zu\r\n%s\r\n$%d\r\n%d\r\n",
+                          strlen(listen_ip), listen_ip, 
+                          (int)snprintf(NULL, 0, "%d", listen_port), listen_port);
+
+    if (send(master_fd, ssync_cmd, cmd_len, 0) < 0) {
+        perror("send SSYNC");
+        goto cleanup;
+    }
+
+    printf("Sent SSYNC command to master\n");
+
+    // 3. 接收master的+OK\r\n回包
+    char resp[32];
+    if (recv_exact(master_fd, resp, 5) < 0) {
+        fprintf(stderr, "Failed to receive OK from master\n");
+        goto cleanup;
+    }
+    
+    if (memcmp(resp, "+OK\r\n", 5) != 0) {
+        fprintf(stderr, "Unexpected response from master: %.5s\n", resp);
+        goto cleanup;
+    }
+
+    printf("Received OK from master\n");
 
 
+    // 4. Accept来自master的三个连接，接收三个snapshot文件
+    const char *snapshot_files[3] = {
+        "data/kvs_array.db",
+        "data/kvs_rbtree.db",
+        "data/kvs_hash.db"
+    };
+
+    for (int i = 0; i < 3; i++) {
+        struct sockaddr_in client_addr;
+        socklen_t addr_len = sizeof(client_addr);
+        
+        snapshot_fds[i] = accept(listen_fd, (struct sockaddr*)&client_addr, &addr_len);
+        if (snapshot_fds[i] < 0) {
+            perror("accept");
+            goto cleanup;
+        }
+
+        printf("Accepted connection %d from %s:%d\n", 
+               i + 1, inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
+
+        if (receive_snapshot_file(snapshot_fds[i], snapshot_files[i]) < 0) {
+            fprintf(stderr, "Failed to receive snapshot file %d\n", i + 1);
+            goto cleanup;
+        }
+
+        printf("Received snapshot file: %s\n", snapshot_files[i]);
+        close(snapshot_fds[i]);
+        snapshot_fds[i] = -1;
+    }
+
+    // 5. 关闭监听和snapshot连接，发送SREADY
     close(listen_fd);
     listen_fd = -1;
 
-    /* 5. 通知 master 快照传输完毕 */
+    const char *sready_cmd = "*1\r\n$6\r\nSREADY\r\n";
-    ctrl_fd = connect_master(master_ip, master_port);
+    if (send(master_fd, sready_cmd, strlen(sready_cmd), 0) < 0) {
-    if (ctrl_fd >= 0) {
+        perror("send SREADY");
-        send_sready(ctrl_fd);
+        goto cleanup;
-        close(ctrl_fd);
-        ctrl_fd = -1;
     }
 
-    /* 6. bootstrap complete */
+    printf("Sent SREADY to master\n");
-    return 0;
+
+    // 6. 接收回包+OK\r\n
+    if (recv_exact(master_fd, resp, 5) < 0) {
+        fprintf(stderr, "Failed to receive final OK from master\n");
+        goto cleanup;
+    }
     
-fail:
+    if (memcmp(resp, "+OK\r\n", 5) != 0) {
-    if (snap_fd   >= 0) close(snap_fd);
+        fprintf(stderr, "Unexpected final response from master: %.5s\n", resp);
-    if (ctrl_fd   >= 0) close(ctrl_fd);
+        goto cleanup;
+    }
+
+    printf("Received final OK from master, bootstrap complete\n");
+    
+    ret = 0;
+
+cleanup:
+    if (master_fd >= 0) close(master_fd);
     if (listen_fd >= 0) close(listen_fd);
-    return -1;
+    for (int i = 0; i < 3; i++) {
+        if (snapshot_fds[i] >= 0) close(snapshot_fds[i]);
+    }
+    
+    return ret;
 }

kvstore.c

@@ -26,11 +26,13 @@ extern mp_pool_t global_mempool;
 #endif
 
 AppConfig global_cfg;
+iouring_ctx_t global_uring_ctx;
 
 extern int global_oplog_fd;
-extern iouring_ctx_t global_uring_ctx;
 
+void __completed_cmd(const uint8_t *cmd, size_t len){
 
+}
 
 int kvs_protocol(struct conn* conn){
 	if (!conn) return -1;
@@ -67,6 +69,9 @@ int kvs_protocol(struct conn* conn){
 
 		int dr = resp_dispatch(&cmd, &val);
 
+		__completed_cmd(p, len);
+
+
         /*
          * 语义建议：
          * - resp_dispatch() 即使返回 -1（比如 unknown command / wrong argc），

kvstore.h

@@ -4,7 +4,7 @@
 #ifndef __KV_STORE_H__
 #define __KV_STORE_H__
 
-
+#include "diskuring/diskuring.h"
 #include <stdio.h>
 #include <string.h>
 #include <stdlib.h>
@@ -82,7 +82,7 @@ int kvs_array_mod_bin(kvs_array_t *inst,
 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_save(iouring_ctx_t *uring, kvs_array_t *inst, const char* filename);
 int kvs_array_load(kvs_array_t *inst, const char* filename);
 
 #else 
@@ -146,7 +146,7 @@ int kvs_rbtree_del(rbtree *inst, const void *key, uint32_t key_len);
 int kvs_rbtree_mod(kvs_rbtree_t *inst, const void *key, uint32_t key_len, const void *value, uint32_t value_len);
 int kvs_rbtree_exist(kvs_rbtree_t *inst, const void *key, uint32_t key_len);
 
-int kvs_rbtree_save(kvs_rbtree_t *inst, const char* filename);
+int kvs_rbtree_save(iouring_ctx_t *uring, kvs_rbtree_t *inst, const char* filename);
 int kvs_rbtree_load(kvs_rbtree_t *inst, const char* filename);
 #else
 #define ENABLE_KEY_CHAR		1
@@ -220,7 +220,7 @@ 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, uint32_t key_len);
 int kvs_hash_count(kvs_hash_t *h);
 
-int kvs_hash_save(kvs_hash_t *inst, const char* filename);
+int kvs_hash_save(iouring_ctx_t *uring, kvs_hash_t *inst, const char* filename);
 int kvs_hash_load(kvs_hash_t *inst, const char* filename);
 #else
 
@@ -279,7 +279,7 @@ extern kvs_rbtree_t global_rbtree;
 extern kvs_hash_t global_hash;
 #endif
 
-void __compeleted_cmd(const uint8_t *cmd, size_t len);
+void __completed_cmd(const uint8_t *cmd, size_t len);
 
 #endif
 

reactor.c

@@ -104,7 +104,7 @@ int event_register(int fd, int event) {
 	memset(conn_list[fd].wbuffer, 0, BUFFER_LENGTH*2);
 	conn_list[fd].wlength = 0;
 
-	conn_list[fd].is_from_master = 0;
+	conn_list[fd].is_stop = 0;
 	
 	set_event(fd, event, 1);
 }
@@ -147,24 +147,24 @@ int recv_cb(int fd) {
 	// printf("avail: %d\n", avail);
 	if (avail <= 0) {
         // 缓冲满了还没解析出来：协议异常或包过大
-        close(fd);
         epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
+        close(fd);
         return 0;
     }
 
 	int count = recv(fd, c->rbuffer + c->rlength, avail, 0);
 	if (count == 0) { // disconnect
 		//printf("client disconnect: %d\n", fd);
-		close(fd);
 
 		epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); // unfinished
+		close(fd);
 
 		return 0;
 	} else if (count < 0) { // 
 
 		printf("count: %d, errno: %d, %s\n", count, errno, strerror(errno));
-		close(fd);
 		epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
+		close(fd);
 
 		return 0;
 	}
@@ -191,8 +191,8 @@ int recv_cb(int fd) {
 #elif ENABLE_KVSTORE
 	int consumed = kvs_request(c);
 	if(consumed < 0){
-		close(fd);
 		epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
+		close(fd);
 		return 0;
 	}
 
@@ -274,8 +274,8 @@ int send_cb(int fd) {
             pthread_mutex_unlock(&c->g_sync_lock);
 
             printf("send fd=%d errno=%d %s\n", fd, e, strerror(e));
-            close(fd);
             epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
+            close(fd);
             return 0;
         }
 
@@ -372,7 +372,7 @@ 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;
+		conn_list[sockfd].is_stop = 0;
 
 		set_event(sockfd, EPOLLIN, 1);
 	}

server.h

@@ -33,7 +33,7 @@ struct conn {
 		RCALLBACK accept_callback;
 	} r_action;
 
-	int is_from_master;
+	int is_stop;
 
 	pthread_mutex_t g_sync_lock;
 

test/test_client.h

@@ -52,7 +52,7 @@ enum {
 	KVS_CMD_HMOD,
 	KVS_CMD_HEXIST,
 
-	KVS_CMD_PSYNC,
+	KVS_CMD_SSYNC,
 	KVS_CMD_SAVE,
     
 	KVS_CMD_COUNT,