From dbf227b8605155e7c8ddebcccab455899b8378a2 Mon Sep 17 00:00:00 2001 From: "zhaozhao.zz" Date: Fri, 31 May 2024 11:37:21 +0800 Subject: [PATCH 1/3] Fix RESP issue about CLUSTER SLOTS cache Signed-off-by: zhaozhao.zz --- src/cluster.c | 24 +++++++++++++----------- src/cluster.h | 2 +- src/cluster_legacy.c | 4 +++- src/networking.c | 3 ++- src/server.h | 4 ++-- 5 files changed, 21 insertions(+), 16 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 71d1cc9124..33c5a0ee1b 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1354,15 +1354,17 @@ void addNodeReplyForClusterSlot(client *c, clusterNode *node, int start_slot, in void clearCachedClusterSlotsResponse(void) { for (connTypeForCaching conn_type = CACHE_CONN_TCP; conn_type < CACHE_CONN_TYPE_MAX; conn_type++) { - if (server.cached_cluster_slot_info[conn_type]) { - sdsfree(server.cached_cluster_slot_info[conn_type]); - server.cached_cluster_slot_info[conn_type] = NULL; + for (int resp = 0; resp <= 3; resp++) { + if (server.cached_cluster_slot_info[conn_type][resp]) { + sdsfree(server.cached_cluster_slot_info[conn_type][resp]); + server.cached_cluster_slot_info[conn_type][resp] = NULL; + } } } } -sds generateClusterSlotResponse(void) { - client *recording_client = createCachedResponseClient(); +sds generateClusterSlotResponse(int resp) { + client *recording_client = createCachedResponseClient(resp); clusterNode *n = NULL; int num_masters = 0, start = -1; void *slot_replylen = addReplyDeferredLen(recording_client); @@ -1392,8 +1394,8 @@ sds generateClusterSlotResponse(void) { return cluster_slot_response; } -int verifyCachedClusterSlotsResponse(sds cached_response) { - sds generated_response = generateClusterSlotResponse(); +int verifyCachedClusterSlotsResponse(sds cached_response, int resp) { + sds generated_response = generateClusterSlotResponse(resp); int is_equal = !sdscmp(generated_response, cached_response); /* Here, we use LL_WARNING so this gets printed when debug assertions are enabled and the system is about to crash. */ if (!is_equal) @@ -1417,12 +1419,12 @@ void clusterCommandSlots(client *c) { if (detectAndUpdateCachedNodeHealth()) clearCachedClusterSlotsResponse(); - sds cached_reply = server.cached_cluster_slot_info[conn_type]; + sds cached_reply = server.cached_cluster_slot_info[conn_type][c->resp]; if (!cached_reply) { - cached_reply = generateClusterSlotResponse(); - server.cached_cluster_slot_info[conn_type] = cached_reply; + cached_reply = generateClusterSlotResponse(c->resp); + server.cached_cluster_slot_info[conn_type][c->resp] = cached_reply; } else { - debugServerAssertWithInfo(c, NULL, verifyCachedClusterSlotsResponse(cached_reply) == 1); + debugServerAssertWithInfo(c, NULL, verifyCachedClusterSlotsResponse(cached_reply, c->resp) == 1); } addReplyProto(c, cached_reply, sdslen(cached_reply)); diff --git a/src/cluster.h b/src/cluster.h index de58486440..03c5100c4b 100644 --- a/src/cluster.h +++ b/src/cluster.h @@ -105,7 +105,7 @@ long long clusterNodeReplOffset(clusterNode *node); clusterNode *clusterLookupNode(const char *name, int length); void clusterReplicateOpenSlots(void); int detectAndUpdateCachedNodeHealth(void); -client *createCachedResponseClient(void); +client *createCachedResponseClient(int resp); void deleteCachedResponseClient(client *recording_client); void clearCachedClusterSlotsResponse(void); diff --git a/src/cluster_legacy.c b/src/cluster_legacy.c index 22fdb20cf7..a4ad0748de 100644 --- a/src/cluster_legacy.c +++ b/src/cluster_legacy.c @@ -1024,7 +1024,9 @@ void clusterInit(void) { server.cluster->mf_end = 0; server.cluster->mf_slave = NULL; for (connTypeForCaching conn_type = CACHE_CONN_TCP; conn_type < CACHE_CONN_TYPE_MAX; conn_type++) { - server.cached_cluster_slot_info[conn_type] = NULL; + for (int resp = 0; resp <= 3; resp++) { + server.cached_cluster_slot_info[conn_type][resp] = NULL; + } } resetManualFailover(); clusterUpdateMyselfFlags(); diff --git a/src/networking.c b/src/networking.c index e062bc3aba..3631eeacad 100644 --- a/src/networking.c +++ b/src/networking.c @@ -332,8 +332,9 @@ sds aggregateClientOutputBuffer(client *c) { * to initiate caching of any command response. * * It needs be paired with `deleteCachedResponseClient` function to stop caching. */ -client *createCachedResponseClient(void) { +client *createCachedResponseClient(int resp) { struct client *recording_client = createClient(NULL); + recording_client->resp = resp; /* Allocating the `conn` allows to prepare the caching client before adding * data to the clients output buffer by `prepareClientToWrite`. */ recording_client->conn = zcalloc(sizeof(connection)); diff --git a/src/server.h b/src/server.h index 249d896d35..eed005e136 100644 --- a/src/server.h +++ b/src/server.h @@ -2068,7 +2068,7 @@ struct valkeyServer { unsigned long long cluster_link_msg_queue_limit_bytes; /* Memory usage limit on individual link msg queue */ int cluster_drop_packet_filter; /* Debug config that allows tactically * dropping packets of a specific type */ - sds cached_cluster_slot_info[CACHE_CONN_TYPE_MAX]; + sds cached_cluster_slot_info[CACHE_CONN_TYPE_MAX][4]; /* Align to RESP3 */ /* Scripting */ mstime_t busy_reply_threshold; /* Script / module timeout in milliseconds */ int pre_command_oom_state; /* OOM before command (script?) was started */ @@ -2725,7 +2725,7 @@ void initSharedQueryBuf(void); client *lookupClientByID(uint64_t id); int authRequired(client *c); void putClientInPendingWriteQueue(client *c); -client *createCachedResponseClient(void); +client *createCachedResponseClient(int resp); void deleteCachedResponseClient(client *recording_client); /* logreqres.c - logging of requests and responses */ From d68d9d578d7e7456f78bd2dfcfd21e892ab3f0b0 Mon Sep 17 00:00:00 2001 From: "zhaozhao.zz" Date: Fri, 31 May 2024 11:38:41 +0800 Subject: [PATCH 2/3] Fix the tls issue about CLUSTER SLOTS cache Signed-off-by: zhaozhao.zz --- src/cluster.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/cluster.c b/src/cluster.c index 33c5a0ee1b..6744075c72 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1415,7 +1415,7 @@ void clusterCommandSlots(client *c) { * 3) node ID * ... continued until done */ - connTypeForCaching conn_type = connIsTLS(c->conn); + connTypeForCaching conn_type = shouldReturnTlsInfo(); if (detectAndUpdateCachedNodeHealth()) clearCachedClusterSlotsResponse(); From eb956a2aba6cbec3830cddb1580729e1612ad8a7 Mon Sep 17 00:00:00 2001 From: "zhaozhao.zz" Date: Tue, 18 Jun 2024 14:23:46 +0800 Subject: [PATCH 3/3] clang format Signed-off-by: zhaozhao.zz --- src/server.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/server.h b/src/server.h index 3d4fee41f9..da0efa15d6 100644 --- a/src/server.h +++ b/src/server.h @@ -2071,7 +2071,7 @@ struct valkeyServer { * dropping packets of a specific type */ /* Debug config that goes along with cluster_drop_packet_filter. When set, the link is closed on packet drop. */ uint32_t debug_cluster_close_link_on_packet_drop : 1; - sds cached_cluster_slot_info[CACHE_CONN_TYPE_MAX][4]; /* Align to RESP3 */ + sds cached_cluster_slot_info[CACHE_CONN_TYPE_MAX][4]; /* Align to RESP3 */ /* Scripting */ mstime_t busy_reply_threshold; /* Script / module timeout in milliseconds */ int pre_command_oom_state; /* OOM before command (script?) was started */