fix unused lambda capture

Backport the time thread code from the enterprise version of KeyDB.

Also fix an issue with clustering where connections are potentially written to after being closed.

pkg/deb/debian/keydb-tools.install

@@ -5,3 +5,4 @@ src/keydb-check-rdb	/usr/bin
 src/keydb-cli		/usr/bin
 src/keydb-server	/usr/bin
 src/keydb-sentinel	/usr/bin
+src/keydb-diagnostic-tool	/usr/bin

pkg/deb/debian_dh9/keydb-tools.install

@@ -5,3 +5,4 @@ src/keydb-check-aof	/usr/bin
 src/keydb-check-rdb	/usr/bin
 src/keydb-cli		/usr/bin
 src/keydb-sentinel	/usr/bin
+src/keydb-diagnostic-tool	/usr/bin

pkg/rpm/generate_rpms.sh

@@ -25,7 +25,13 @@ mkdir -p $DIR/keydb_build/keydb_rpm/var/log/keydb
 
 # move binaries to bin
 rm $DIR/keydb_build/keydb_rpm/usr/bin/*
-cp $DIR/../../src/keydb-* $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-server $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-sentinel $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-cli $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-benchmark $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-check-aof $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-check-rdb $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-diagnostic-tool $DIR/keydb_build/keydb_rpm/usr/bin/
 
 # update spec file with build info
 sed -i '2d' $DIR/keydb_build/keydb.spec

src/blocked.cpp

@@ -592,7 +592,6 @@ void handleClientsBlockedOnKeys(void) {
              * lookup, invalidating the first one.
              * See https://github.com/redis/redis/pull/6554. */
             serverTL->fixed_time_expire++;
-            updateCachedTime(0);
 
             /* Serve clients blocked on the key. */
             robj *o = lookupKeyWrite(rl->db,rl->key);

src/cluster.cpp

@@ -2447,7 +2447,10 @@ void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {
     if (sdslen(link->sndbuf) == 0 && msglen != 0)
     {
         aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [link] {
-            connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1);
+            /* The connection could be timed out before this posted function executes (thanks to TCP keepalive).
+             * So check that the connection is still there before setting the write handler, otherwise you segfault */
+            if (link->conn != nullptr)
+                connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1);
         });
     }
 

src/module.cpp

@@ -4289,7 +4289,14 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
         if (!(flags & REDISMODULE_ARGV_NO_REPLICAS))
             call_flags |= CMD_CALL_PROPAGATE_REPL;
     }
+
+    {
+    aeAcquireLock();
+    std::unique_lock<fastlock> ul(c->lock);
     call(c,call_flags);
+    aeReleaseLock();
+    }
+
     g_pserver->replication_allowed = prev_replication_allowed;
 
     serverAssert((c->flags & CLIENT_BLOCKED) == 0);

src/networking.cpp

@@ -1321,7 +1321,7 @@ void acceptOnThread(connection *conn, int flags, char *cip)
             szT = (char*)zmalloc(NET_IP_STR_LEN, MALLOC_LOCAL);
             memcpy(szT, cip, NET_IP_STR_LEN);
         }
-        int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT, fBootLoad] {
+        int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT] {
             connMarshalThread(conn);
             acceptCommonHandler(conn,flags,szT,ielTarget);
             rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
@@ -1942,7 +1942,9 @@ void ProcessPendingAsyncWrites()
                 bool fResult = c->postFunction([](client *c) {
                     c->fPendingAsyncWriteHandler = false;
                     clientInstallWriteHandler(c);
+                    c->lock.unlock();
                     handleClientsWithPendingWrites(c->iel, g_pserver->aof_state);
+                    c->lock.lock();
                 }, false);
 
                 if (!fResult)
@@ -3867,10 +3869,6 @@ void processEventsWhileBlocked(int iel) {
     listNode *ln;
     listRewind(g_pserver->clients, &li);
 
-    /* Update our cached time since it is used to create and update the last
-     * interaction time with clients and for other important things. */
-    updateCachedTime(0);
-
     // All client locks must be acquired *after* the global lock is reacquired to prevent deadlocks
     //  so unlock here, and save them for reacquisition later
     while ((ln = listNext(&li)) != nullptr)

src/redis-benchmark.cpp

@@ -1984,7 +1984,6 @@ int main(int argc, const char **argv) {
     } while(config.loop);
     zfree(data);
 
-    zfree(data);
     if (config.redis_config != NULL) freeRedisConfig(config.redis_config);
 
     return 0;

src/server.cpp

@@ -63,6 +63,7 @@
 #include <algorithm>
 #include <uuid/uuid.h>
 #include <mutex>
+#include <condition_variable>
 #include "aelocker.h"
 #include "motd.h"
 #include "t_nhash.h"
@@ -94,6 +95,10 @@ struct redisServer server; /* Server global state */
 redisServer *g_pserver = &GlobalHidden::server;
 struct redisServerConst cserver;
 __thread struct redisServerThreadVars *serverTL = NULL;   // thread local server vars
+std::mutex time_thread_mutex;
+std::condition_variable time_thread_cv;
+int sleeping_threads = 0;
+void wakeTimeThread();
 
 /* Our command table.
  *
@@ -2007,7 +2012,7 @@ void databasesCron(bool fMainThread) {
  * info or not using the 'update_daylight_info' argument. Normally we update
  * such info only when calling this function from serverCron() but not when
  * calling it from call(). */
-void updateCachedTime(int update_daylight_info) {
+void updateCachedTime() {
     long long t = ustime();
     __atomic_store(&g_pserver->ustime, &t, __ATOMIC_RELAXED);
     t /= 1000;
@@ -2020,12 +2025,10 @@ void updateCachedTime(int update_daylight_info) {
      * context is safe since we will never fork() while here, in the main
      * thread. The logging function will call a thread safe version of
      * localtime that has no locks. */
-    if (update_daylight_info) {
-        struct tm tm;
-        time_t ut = g_pserver->unixtime;
-        localtime_r(&ut,&tm);
-        __atomic_store(&g_pserver->daylight_active, &tm.tm_isdst, __ATOMIC_RELAXED);
-    }
+    struct tm tm;
+    time_t ut = g_pserver->unixtime;
+    localtime_r(&ut,&tm);
+    __atomic_store(&g_pserver->daylight_active, &tm.tm_isdst, __ATOMIC_RELAXED);
 }
 
 void checkChildrenDone(void) {
@@ -2152,9 +2155,6 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
      * handler if we don't return here fast enough. */
     if (g_pserver->watchdog_period) watchdogScheduleSignal(g_pserver->watchdog_period);
 
-    /* Update the time cache. */
-    updateCachedTime(1);
-
     g_pserver->hz = g_pserver->config_hz;
     /* Adapt the g_pserver->hz value to the number of configured clients. If we have
      * many clients, we want to call serverCron() with an higher frequency. */
@@ -2412,7 +2412,6 @@ int serverCronLite(struct aeEventLoop *eventLoop, long long id, void *clientData
 
 void blockingOperationStarts() {
     if(!g_pserver->blocking_op_nesting++){
-        updateCachedTime(0);
         g_pserver->blocked_last_cron = g_pserver->mstime;
     }
 }
@@ -2623,6 +2622,14 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
     locker.disarm();
     if (!fSentReplies)
         handleClientsWithPendingWrites(iel, aof_state);
+
+    // Scope lock_guard
+    {
+        std::lock_guard<std::mutex> lock(time_thread_mutex);
+        sleeping_threads++;
+        serverAssert(sleeping_threads <= cserver.cthreads);
+    }
+    
     /* Determine whether the modules are enabled before sleeping, and use that result
        both here, and after wakeup to avoid double acquire or release of the GIL */
     serverTL->modulesEnabledThisAeLoop = !!moduleCount();
@@ -2642,6 +2649,7 @@ void afterSleep(struct aeEventLoop *eventLoop) {
        Don't check here that modules are enabled, rather use the result from beforeSleep
        Otherwise you may double acquire the GIL and cause deadlocks in the module */
     if (!ProcessingEventsWhileBlocked) {
+        wakeTimeThread();
         if (serverTL->modulesEnabledThisAeLoop) moduleAcquireGIL(TRUE /*fServerThread*/);
     }
 }
@@ -2839,7 +2847,7 @@ void initMasterInfo(redisMaster *master)
 void initServerConfig(void) {
     int j;
 
-    updateCachedTime(true);
+    updateCachedTime();
     getRandomHexChars(g_pserver->runid,CONFIG_RUN_ID_SIZE);
     g_pserver->runid[CONFIG_RUN_ID_SIZE] = '\0';
     changeReplicationId();
@@ -4006,7 +4014,6 @@ void call(client *c, int flags) {
     prev_err_count = g_pserver->stat_total_error_replies;
     g_pserver->fixed_time_expire++;
 
-    updateCachedTime(0);
     incrementMvccTstamp();
     elapsedStart(&call_timer);
     try {
@@ -6534,6 +6541,35 @@ void OnTerminate()
     serverPanic("std::teminate() called");
 }
 
+void wakeTimeThread() {
+    updateCachedTime();
+    std::lock_guard<std::mutex> lock(time_thread_mutex);
+    if (sleeping_threads >= cserver.cthreads)
+        time_thread_cv.notify_one();
+    sleeping_threads--;
+    serverAssert(sleeping_threads >= 0);
+}
+
+void *timeThreadMain(void*) {
+    timespec delay;
+    delay.tv_sec = 0;
+    delay.tv_nsec = 100;
+    while (true) {
+        {
+            std::unique_lock<std::mutex> lock(time_thread_mutex);
+            if (sleeping_threads >= cserver.cthreads) {
+                time_thread_cv.wait(lock);
+            }
+        }
+        updateCachedTime();
+#if defined(__APPLE__)
+        nanosleep(&delay, nullptr);
+#else
+        clock_nanosleep(CLOCK_MONOTONIC, 0, &delay, NULL);
+#endif
+    }
+}
+
 void *workerThreadMain(void *parg)
 {
     int iel = (int)((int64_t)parg);
@@ -6930,6 +6966,8 @@ int main(int argc, char **argv) {
     setOOMScoreAdj(-1);
     serverAssert(cserver.cthreads > 0 && cserver.cthreads <= MAX_EVENT_LOOPS);
 
+    pthread_create(&cserver.time_thread_id, nullptr, timeThreadMain, nullptr);
+
     pthread_attr_t tattr;
     pthread_attr_init(&tattr);
     pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB

src/server.h

@@ -1536,6 +1536,7 @@ struct redisServerConst {
     pid_t pid;                  /* Main process pid. */
     time_t stat_starttime;          /* Server start time */
     pthread_t main_thread_id;         /* Main thread id */
+    pthread_t time_thread_id;
     char *configfile;           /* Absolute config file path, or NULL */
     char *executable;           /* Absolute executable file path. */
     char **exec_argv;           /* Executable argv vector (copy). */
@@ -2637,7 +2638,7 @@ void resetErrorTableStats(void);
 void adjustOpenFilesLimit(void);
 void incrementErrorCount(const char *fullerr, size_t namelen);
 void closeListeningSockets(int unlink_unix_socket);
-void updateCachedTime(int update_daylight_info);
+void updateCachedTime(void);
 void resetServerStats(void);
 void activeDefragCycle(void);
 unsigned int getLRUClock(void);