From 50e8cee88d57e2db35be3696dc589e4b32ba7cc5 Mon Sep 17 00:00:00 2001 From: Victor Date: Fri, 20 Dec 2024 05:47:33 +0000 Subject: [PATCH] [WIP] ping with swap_info --- src/ctrip_swap.h | 4 ++- src/ctrip_swap_repl.c | 72 +++++++++++++++++++------------------------ src/server.c | 13 +++++--- 3 files changed, 44 insertions(+), 45 deletions(-) diff --git a/src/ctrip_swap.h b/src/ctrip_swap.h index 5d2e2ca3fb5..a21960693ad 100644 --- a/src/ctrip_swap.h +++ b/src/ctrip_swap.h @@ -1997,7 +1997,9 @@ void swapInfoCommand(client *c); void swapBuildSwapInfoSstAgeLimitCmd(robj *argv[3], long long sst_age_limit); void swapDestorySwapInfoSstAgeLimitCmd(robj *argv[3]); -void swapPropagateSwapInfo(int argc, robj **argv); +void swapPropagateSwapInfo(sds ping_argv); +sds swapEncodeSwapInfo(int argc, sds *argv); +sds *swapDecodeSwapInfo(sds argv); /* Repl */ int submitReplClientRequests(client *c); diff --git a/src/ctrip_swap_repl.c b/src/ctrip_swap_repl.c index 44ba38b148c..06b4be87ed4 100644 --- a/src/ctrip_swap_repl.c +++ b/src/ctrip_swap_repl.c @@ -392,59 +392,51 @@ sds genSwapReplInfoString(sds info) { return info; } -bool isSwapInfoSupported(void) { +/* SWAP.INFO SST-AGE-LIMIT */ +void swapBuildSwapInfoSstAgeLimitCmd(sds *argv[3], long long sst_age_limit) { - if (server.swap_swap_info_supported == SWAP_INFO_SUPPORTED_YES) return true; - if (server.swap_swap_info_supported == SWAP_INFO_SUPPORTED_NO) return false; + argv[0] = shared.swap_info->ptr; + argv[1] = shared.sst_age_limit->ptr; + argv[2] = sdsfromlonglong(sst_age_limit); +} - /* SWAP_INFO_SUPPORTED_AUTO */ - /* depends on capa of all slaves, - * once there is one slave without capa of swap.info, return false. */ +void swapDestorySwapInfoSstAgeLimitCmd(sds *argv[3]) { + sdsfree(argv[2]); +} - listNode *ln; - listIter li; +/* The swap.info argv, propagate system info to slave with ping command. + * SWAP.INFO [ [value] [opt] ...] + * + * subcommand supported: + * SWAP.INFO SST-AGE-LIMIT */ +sds swapEncodeSwapInfo(int argc, sds *argv) { - listRewind(server.slaves,&li); - while((ln = listNext(&li))) { - client *slave = ln->value; + sds ping_argv = sdsempty(); - if (!(slave->slave_capa & SLAVE_CAPA_SWAP_INFO)) { - return false; - } + + for (int i = 0; i< argc; i++) { + ping_argv = sdscat(ping_argv, argv); + ping_argv = sdscat(ping_argv, " "); /* spilt by space */ } - return true; + + return ping_argv; } -/* SWAP.INFO SST-AGE-LIMIT */ -void swapBuildSwapInfoSstAgeLimitCmd(robj *argv[3], long long sst_age_limit) { +sds *swapDecodeSwapInfo(sds argv) { - argv[0] = shared.swap_info; - argv[1] = shared.sst_age_limit; - argv[2] = createStringObjectFromLongLong(sst_age_limit); -} -void swapDestorySwapInfoSstAgeLimitCmd(robj *argv[3]) { - decrRefCount(argv[2]); -} + /* SWAP.INFO SST-AGE-LIMIT */ -/* The swap.info command, propagate system info to slave. - * SWAP.INFO [ [value] [opt] ...] - * - * subcommand supported: - * SWAP.INFO SST-AGE-LIMIT */ -void swapPropagateSwapInfo(int argc, robj **argv) { + return NULL ; +} - if (!isSwapInfoSupported()) return; +void swapPropagateSwapInfo(sds ping_argv) { - if (argc < 2) { - return; - } else if (argc == 3 && !strcasecmp(argv[1]->ptr,"SST-AGE-LIMIT")) { - /* SWAP.INFO SST-AGE-LIMIT */ - goto propagate; - } - return; + robj *argv[2]; -propagate: - replicationFeedSlaves(server.slaves,0,argv,argc); + argv[0] = shared.ping; + argv[1] = createStringObject(ping_argv, sdslen(ping_argv)); + + replicationFeedSlaves(server.slaves,0,argv,2); return; } diff --git a/src/server.c b/src/server.c index 779d0a9a58e..a01f4843405 100644 --- a/src/server.c +++ b/src/server.c @@ -2486,9 +2486,10 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { run_with_period(1000*(int)server.swap_swap_info_slave_period) { /* propagate sst age limit */ if (server.swap_ttl_compact_enabled) { - robj *argv[3]; + sds *argv[3]; swapBuildSwapInfoSstAgeLimitCmd(argv, server.swap_ttl_compact_ctx->expire_stats->sst_age_limit); - swapPropagateSwapInfo(3, argv); + sds swap_info = swapEncodeSwapInfo(3, argv); + swapPropagateSwapInfo(swap_info); swapDestorySwapInfoSstAgeLimitCmd(argv); } } @@ -4900,10 +4901,14 @@ void pingCommand(client *c) { else addReplyBulk(c,c->argv[1]); } else { - if (c->argc == 1) + if (c->argc == 1) { addReply(c,shared.pong); - else + } else { + sds *swap_info = swapDecodeSwapInfo(c->argv[1]->ptr); + /* apply swap info ... */ + /* ... */ addReplyBulk(c,c->argv[1]); + } } }