Merge remote-tracking branch 'valkey/unstable' into hashset

valkey-io · Oct 1, 2024 · 2b77a02 · 2b77a02
2 parents 9c34fe7 + 613e4e0
commit 2b77a02
Show file tree

Hide file tree

Showing 12 changed files with 51 additions and 16 deletions.
diff --git a/src/ae.c b/src/ae.c
@@ -363,7 +363,7 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
         }
 
         if (te->when <= now) {
-            int retval;
+            long long retval;
 
             id = te->id;
             te->refcount++;

diff --git a/src/ae.h b/src/ae.h
@@ -67,7 +67,7 @@ struct aeEventLoop;
 
 /* Types and data structures */
 typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
-typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
+typedef long long aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
 typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);
 typedef void aeBeforeSleepProc(struct aeEventLoop *eventLoop);
 typedef void aeAfterSleepProc(struct aeEventLoop *eventLoop, int numevents);

diff --git a/src/evict.c b/src/evict.c
@@ -447,7 +447,7 @@ int overMaxmemoryAfterAlloc(size_t moremem) {
  * eviction cycles until the "maxmemory" condition has resolved or there are no
  * more evictable items.  */
 static int isEvictionProcRunning = 0;
-static int evictionTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData) {
+static long long evictionTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     UNUSED(eventLoop);
     UNUSED(id);
     UNUSED(clientData);

diff --git a/src/module.c b/src/module.c
@@ -9113,7 +9113,7 @@ typedef struct ValkeyModuleTimer {
 
 /* This is the timer handler that is called by the main event loop. We schedule
  * this timer to be called when the nearest of our module timers will expire. */
-int moduleTimerHandler(struct aeEventLoop *eventLoop, long long id, void *clientData) {
+long long moduleTimerHandler(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     UNUSED(eventLoop);
     UNUSED(id);
     UNUSED(clientData);

diff --git a/src/notify.c b/src/notify.c
@@ -30,7 +30,7 @@
 #include "server.h"
 
 /* This file implements keyspace events notification via Pub/Sub and
- * described at https://redis.io/topics/notifications. */
+ * described at https://valkey.io/topics/notifications */
 
 /* Turn a string representing notification classes into an integer
  * representing notification classes flags xored.

diff --git a/src/rdma.c b/src/rdma.c
@@ -672,7 +672,7 @@ static void connRdmaEventHandler(struct aeEventLoop *el, int fd, void *clientDat
     }
 }
 
-static int rdmaKeepaliveTimeProc(struct aeEventLoop *el, long long id, void *clientData) {
+static long long rdmaKeepaliveTimeProc(struct aeEventLoop *el, long long id, void *clientData) {
     struct rdma_cm_id *cm_id = clientData;
     RdmaContext *ctx = cm_id->context;
     connection *conn = ctx->conn;
@@ -1501,6 +1501,26 @@ static int rdmaServer(char *err, int port, char *bindaddr, int af, rdma_listener
     return ret;
 }
 
+static int connRdmaIsLocal(connection *conn) {
+    rdma_connection *rdma_conn = (rdma_connection *)conn;
+    struct sockaddr *laddr = rdma_get_local_addr(rdma_conn->cm_id);
+    struct sockaddr *raddr = rdma_get_peer_addr(rdma_conn->cm_id);
+    struct sockaddr_in *lsa4, *rsa4;
+    struct sockaddr_in6 *lsa6, *rsa6;
+
+    if (laddr->sa_family == AF_INET) {
+        lsa4 = (struct sockaddr_in *)laddr;
+        rsa4 = (struct sockaddr_in *)raddr;
+        return !memcmp(&lsa4->sin_addr, &rsa4->sin_addr, sizeof(lsa4->sin_addr));
+    } else if (laddr->sa_family == AF_INET6) {
+        lsa6 = (struct sockaddr_in6 *)laddr;
+        rsa6 = (struct sockaddr_in6 *)raddr;
+        return !memcmp(&lsa6->sin6_addr, &rsa6->sin6_addr, sizeof(lsa6->sin6_addr));
+    }
+
+    return -1;
+}
+
 int connRdmaListen(connListener *listener) {
     int j, ret;
     char **bindaddr = listener->bindaddr;
@@ -1673,6 +1693,7 @@ static ConnectionType CT_RDMA = {
     .ae_handler = connRdmaEventHandler,
     .accept_handler = connRdmaAcceptHandler,
     //.cluster_accept_handler = NULL,
+    .is_local = connRdmaIsLocal,
     .listen = connRdmaListen,
     .addr = connRdmaAddr,
 

diff --git a/src/replication.c b/src/replication.c
@@ -2738,8 +2738,10 @@ static void fullSyncWithPrimary(connection *conn) {
 
 error:
     sdsfree(err);
-    connClose(conn);
-    server.repl_transfer_s = NULL;
+    if (server.repl_transfer_s) {
+        connClose(server.repl_transfer_s);
+        server.repl_transfer_s = NULL;
+    }
     if (server.repl_rdb_transfer_s) {
         connClose(server.repl_rdb_transfer_s);
         server.repl_rdb_transfer_s = NULL;

diff --git a/src/script_lua.c b/src/script_lua.c
@@ -361,7 +361,7 @@ static void redisProtocolToLuaType_Map(struct ReplyParser *parser, void *ctx, si
         }
         lua_newtable(lua);
         lua_pushstring(lua, "map");
-        lua_newtable(lua);
+        lua_createtable(lua, 0, len);
     }
     for (size_t j = 0; j < len; j++) {
         parseReply(parser, lua);
@@ -383,7 +383,7 @@ static void redisProtocolToLuaType_Set(struct ReplyParser *parser, void *ctx, si
         }
         lua_newtable(lua);
         lua_pushstring(lua, "set");
-        lua_newtable(lua);
+        lua_createtable(lua, 0, len);
     }
     for (size_t j = 0; j < len; j++) {
         parseReply(parser, lua);
@@ -412,7 +412,7 @@ static void redisProtocolToLuaType_Array(struct ReplyParser *parser, void *ctx,
              * to push elements to the stack. On failure, exit with panic. */
             serverPanic("lua stack limit reach when parsing server.call reply");
         }
-        lua_newtable(lua);
+        lua_createtable(lua, len, 0);
     }
     for (size_t j = 0; j < len; j++) {
         if (lua) lua_pushnumber(lua, j + 1);
@@ -1534,7 +1534,7 @@ void luaRegisterServerAPI(lua_State *lua) {
 static void luaCreateArray(lua_State *lua, robj **elev, int elec) {
     int j;
 
-    lua_newtable(lua);
+    lua_createtable(lua, elec, 0);
     for (j = 0; j < elec; j++) {
         lua_pushlstring(lua, (char *)elev[j]->ptr, sdslen(elev[j]->ptr));
         lua_rawseti(lua, -2, j + 1);

diff --git a/src/server.c b/src/server.c
@@ -1273,7 +1273,7 @@ void cronUpdateMemoryStats(void) {
  * a macro is used: run_with_period(milliseconds) { .... }
  */
 
-int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
+long long serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     int j;
     UNUSED(eventLoop);
     UNUSED(id);

diff --git a/src/valkey-benchmark.c b/src/valkey-benchmark.c
@@ -195,7 +195,7 @@ static redisContext *getRedisContext(const char *ip, int port, const char *hosts
 static void freeServerConfig(serverConfig *cfg);
 static int fetchClusterSlotsConfiguration(client c);
 static void updateClusterSlotsConfiguration(void);
-int showThroughput(struct aeEventLoop *eventLoop, long long id, void *clientData);
+static long long showThroughput(struct aeEventLoop *eventLoop, long long id, void *clientData);
 
 /* Dict callbacks */
 static uint64_t dictSdsHash(const void *key);
@@ -1604,7 +1604,7 @@ int parseOptions(int argc, char **argv) {
     exit(exit_status);
 }
 
-int showThroughput(struct aeEventLoop *eventLoop, long long id, void *clientData) {
+long long showThroughput(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     UNUSED(eventLoop);
     UNUSED(id);
     benchmarkThread *thread = (benchmarkThread *)clientData;

diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl
@@ -15,6 +15,7 @@ source tests/support/util.tcl
 
 set dir [pwd]
 set ::all_tests []
+set ::cluster_all_test []
 set ::module_api_all_tests []
 
 set test_dirs {
@@ -32,6 +33,11 @@ foreach test_dir $test_dirs {
     }
 }
 
+set cluster_test_dir unit/cluster
+foreach file [glob -nocomplain $dir/tests/$cluster_test_dir/*.tcl] {
+   lappend ::cluster_all_tests $cluster_test_dir/[file root [file tail $file]]
+}
+
 set moduleapi_test_dir unit/moduleapi
 foreach file [glob -nocomplain $dir/tests/$moduleapi_test_dir/*.tcl] {
    lappend ::module_api_all_tests $moduleapi_test_dir/[file root [file tail $file]]
@@ -527,7 +533,7 @@ proc the_end {} {
     }
 }
 
-# The client is not even driven (the test server is instead) as we just need
+# The client is not event driven (the test server is instead) as we just need
 # to read the command, execute, reply... all this in a loop.
 proc test_client_main server_port {
     set ::test_server_fd [socket localhost $server_port]
@@ -557,6 +563,7 @@ proc send_data_packet {fd status data {elapsed 0}} {
 
 proc print_help_screen {} {
     puts [join {
+        "--cluster          Run the cluster tests, by default cluster tests run along with all tests."
         "--moduleapi        Run the module API tests, this option should only be used in runtest-moduleapi which will build the test module."
         "--valgrind         Run the test over valgrind."
         "--durable          suppress test crashes and keep running"
@@ -614,6 +621,8 @@ for {set j 0} {$j < [llength $argv]} {incr j} {
             }
         }
         incr j
+    } elseif {$opt eq {--cluster}} {
+        set ::all_tests $::cluster_all_tests
     } elseif {$opt eq {--moduleapi}} {
         set ::all_tests $::module_api_all_tests
     } elseif {$opt eq {--config}} {

diff --git a/tests/unit/aofrw.tcl b/tests/unit/aofrw.tcl
@@ -9,6 +9,9 @@ start_server {tags {"aofrw external:skip logreqres:skip"} overrides {save {}}} {
     foreach rdbpre {yes no} {
         r config set aof-use-rdb-preamble $rdbpre
         test "AOF rewrite during write load: RDB preamble=$rdbpre" {
+            # Start with an empty db
+            r flushall
+
             # Start a write load for 10 seconds
             set master [srv 0 client]
             set master_host [srv 0 host]