Backported time thread code from enterprise

pkg/deb/debian/control

@@ -6,7 +6,6 @@ Build-Depends:
  debhelper (>= 9~),
  dpkg-dev (>= 1.17.5),
  systemd,
- libsystemd-dev <!nocheck>,
  procps <!nocheck>,
  build-essential <!nocheck>,
  tcl <!nocheck>,

pkg/deb/debian/copyright

@@ -1,94 +1,16 @@
 Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
 Upstream-Contact: John Sully <john@eqalpha.com>
-Upstream-Name: KeyDB
-Source: https://github.com/EQ-Alpha/KeyDB
+Upstream-Name: keydb-pro
+Source: https://github.com/JohnSully/KeyDB-Pro
 
 Files: *
-Copyright: 
- © 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
- © 2019-2021 John Sully
- © 2019-2021 EQ Alpha Technology Ltd.
+Copyright: © 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
+Copyright © 2019, John Sully
+License: Proprietary
 
-License: BSD-3-clause
 
-Files:
- src/rio.*
- src/t_zset.c
- src/ziplist.h
- src/intset.*
- src/redis-check-aof.c
- deps/hiredis/*
- deps/linenoise/*
-Copyright:
- © 2009-2012 Pieter Noordhuis <pcnoordhuis@gmail.com>
- © 2009-2012 Salvatore Sanfilippo <antirez@gmail.com>
-License: BSD-3-clause
-
-Files:
- src/lzf.h
- src/lzfP.h
- src/lzf_d.c
- src/lzf_c.c
-Copyright:
- © 2000-2007 Marc Alexander Lehmann <schmorp@schmorp.de>
- © 2009-2012 Salvatore Sanfilippo <antirez@gmail.com>
-License: BSD-2-clause
-
-Files: src/setproctitle.c
-Copyright:
- © 2010 William Ahern
- © 2013 Salvatore Sanfilippo
- © 2013 Stam He
-License: BSD-3-clause
-
-Files: src/ae_evport.c
-Copyright: © 2012 Joyent, Inc.
-License: BSD-3-clause
-
-Files: src/ae_kqueue.c
-Copyright: © 2009 Harish Mallipeddi <harish.mallipeddi@gmail.com>
-License: BSD-3-clause
-
-Files: utils/install_server.sh
-Copyright: © 2011 Dvir Volk <dvirsk@gmail.com>
-License: BSD-3-clause
-
-Files: deps/jemalloc/*
-Copyright:
- © 2002-2012 Jason Evans <jasone@canonware.com>
- © 2007-2012 Mozilla Foundation
- © 2009-2012 Facebook, Inc.
-License: BSD-3-clause
-
-Files: src/pqsort.*
-Copyright: © 1992-1993 The Regents of the University of California
-License: BSD-3-clause
-
-Files: deps/lua/*
-Copyright: © 1994-2012 Lua.org, PUC-Ri
-License: MIT
- Permission is hereby granted, free of charge, to any person obtaining a copy
- of this software and associated documentation files (the "Software"), to deal
- in the Software without restriction, including without limitation the rights
- to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- copies of the Software, and to permit persons to whom the Software is
- furnished to do so, subject to the following conditions:
- .
- The above copyright notice and this permission notice shall be included in
- all copies or substantial portions of the Software.
- .
- THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
- THE SOFTWARE.
-
-Files: pkg/deb/debian/*
-Copyright:
- © 2009 Chris Lamb <lamby@debian.org>
- © 2020-2021 EQ Alpha Technology Ltd. <support@eqalpha.com>
+Files: debian/*
+Copyright: © 2009 Chris Lamb <lamby@debian.org>
 License: BSD-3-clause
 
 License: BSD-2-clause

pkg/deb/debian/rules

@@ -3,7 +3,6 @@
 include /usr/share/dpkg/buildflags.mk
 
 export BUILD_TLS=yes
-export USE_SYSTEMD=yes
 export CFLAGS CPPFLAGS LDFLAGS
 export DEB_BUILD_MAINT_OPTIONS = hardening=+all
 export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)

pkg/deb/debian_dh9/control

@@ -6,7 +6,6 @@ Build-Depends:
  debhelper (>= 9~),
  dpkg-dev (>= 1.17.5),
  systemd,
- libsystemd-dev <!nocheck>,
  procps <!nocheck>,
  build-essential <!nocheck>,
  tcl <!nocheck>,

pkg/deb/debian_dh9/copyright

@@ -1,14 +1,11 @@
 Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
 Upstream-Contact: John Sully <john@eqalpha.com>
-Upstream-Name: KeyDB
-Source: https://github.com/EQ-Alpha/KeyDB
+Upstream-Name: keydb
+Source: https://github.com/JohnSully/KeyDB
 
 Files: *
-Copyright: 
- © 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
- © 2019-2021 John Sully
- © 2019-2021 EQ Alpha Technology Ltd.
-
+Copyright: © 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
+Copyright © 2019, John Sully
 License: BSD-3-clause
 
 Files:
@@ -85,10 +82,8 @@ License: MIT
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  THE SOFTWARE.
 
-Files: pkg/deb/debian_dh9/*
-Copyright: 
- © 2009 Chris Lamb <lamby@debian.org>
- © 2020-2021 EQ Alpha Technology Ltd. <support@eqalpha.com>
+Files: debian/*
+Copyright: © 2009 Chris Lamb <lamby@debian.org>
 License: BSD-3-clause
 
 License: BSD-2-clause

pkg/deb/debian_dh9/rules

@@ -9,7 +9,6 @@ include /usr/share/dpkg/buildflags.mk
 #LUA_LDFLAGS = $(addprefix -llua5.1-,$(LUA_LIBS_DEBIAN)) $(addprefix ../deps/lua/src/,$(LUA_OBJECTS))
 
 export BUILD_TLS=yes
-export USE_SYSTEMD=yes
 export CFLAGS CPPFLAGS LDFLAGS
 export DEB_BUILD_MAINT_OPTIONS = hardening=+all
 export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)

src/acl.cpp

@@ -223,18 +223,12 @@ uint64_t ACLGetCommandCategoryFlagByName(const char *name) {
     return 0; /* No match. */
 }
 
-/* Method for pattern comparison used for the user->patterns list
+/* Method for passwords/pattern comparison used for the user->passwords list
  * so that we can search for items with listSearchKey(). */
 int ACLListMatchSds(void *a, void *b) {
     return sdscmp((sds)a,(sds)b) == 0;
 }
 
-/* Method for password comparison used for the user->passwords list
- * Like the above, but uses a time independant compare for security reasons */
-int ACLListMatchSdsSecure(void *a, void* b) {
-    return time_independent_strcmp((sds)a,(sds)b) == 0;
-}
-
 /* Method to free list elements from ACL users password/patterns lists. */
 void ACLListFreeSds(const void *item) {
     sdsfree((sds)item);
@@ -259,7 +253,7 @@ user *ACLCreateUser(const char *name, size_t namelen) {
     u->passwords = listCreate();
     u->patterns = listCreate();
     u->channels = listCreate();
-    listSetMatchMethod(u->passwords,ACLListMatchSdsSecure);
+    listSetMatchMethod(u->passwords,ACLListMatchSds);
     listSetFreeMethod(u->passwords,ACLListFreeSds);
     listSetDupMethod(u->passwords,ACLListDupSds);
     listSetMatchMethod(u->patterns,ACLListMatchSds);

src/bitops.cpp

@@ -405,10 +405,6 @@ void printBits(unsigned char *p, unsigned long count) {
 #define BITFIELDOP_SET 1
 #define BITFIELDOP_INCRBY 2
 
-/* Make sure we don't bit shift too many bits at a time.
- * Even this amount is probably way too large. */
-#define BITOP_SHIFT_MAX (1<<30)
-
 /* This helper function used by GETBIT / SETBIT parses the bit offset argument
  * making sure an error is returned if it is negative or if it overflows
  * Redis 512 MB limit for the string value or more (server.proto_max_bulk_len).
@@ -646,11 +642,6 @@ void bitopCommand(client *c) {
             addReplyError(c, "BITOP SHIFT's last parameter must be an integer");
             return;
         }
-        if (shift > BITOP_SHIFT_MAX) {
-            addReplyError(c, "BITOP SHIFT value is too large.");
-            return;
-        }
-
         if (op == BITOP_RSHIFT)
             shift = -shift;
     }
@@ -691,7 +682,7 @@ void bitopCommand(client *c) {
 
     if (fShiftOp)
     {
-        long long newlen = (long long)maxlen + shift/CHAR_BIT;
+        long newlen = (long)maxlen + shift/CHAR_BIT;
         if (shift > 0 && (shift % CHAR_BIT) != 0)
             newlen++;
 
@@ -703,14 +694,14 @@ void bitopCommand(client *c) {
             res = (unsigned char*) sdsnewlen(NULL,newlen);
             if (shift >= 0)
             {   // left shift
-                long long byteoffset = shift/CHAR_BIT;
+                long byteoffset = shift/CHAR_BIT;
                 memset(res, 0, byteoffset);
-                long long srcLen = newlen - byteoffset - ((shift % CHAR_BIT) ? 1 : 0);
+                long srcLen = newlen - byteoffset - ((shift % CHAR_BIT) ? 1 : 0);
 
                 // now the bitshift+copy
                 unsigned bitshift = shift % CHAR_BIT;
                 unsigned char carry = 0;
-                for (long long iSrc = 0; iSrc < srcLen; ++iSrc)
+                for (long iSrc = 0; iSrc < srcLen; ++iSrc)
                 {
                     res[byteoffset+iSrc] = (src[0][iSrc] << bitshift) | carry;
                     carry = src[0][iSrc] >> (CHAR_BIT - bitshift);
@@ -720,14 +711,14 @@ void bitopCommand(client *c) {
             } 
             else 
             {   // right shift
-                long long byteoffset = -shift/CHAR_BIT;
+                long byteoffset = -shift/CHAR_BIT;
                 unsigned bitshift = -shift % CHAR_BIT;
                 if (bitshift)
                     ++byteoffset;
                 res[0] = (src[0][byteoffset] << (CHAR_BIT-bitshift));
                 if (byteoffset > 0)
                     res[0] |= (src[0][byteoffset-1] >> bitshift);
-                for (long long idx = 1; idx < newlen; ++idx)
+                for (long idx = 1; idx < newlen; ++idx)
                 {
                     res[idx] = (src[0][byteoffset+idx] << (CHAR_BIT-bitshift)) | (src[0][byteoffset+idx-1] >> bitshift);
                 }
@@ -809,8 +800,8 @@ void bitopCommand(client *c) {
                     }
                 }
             }
-            #endif
         }
+        #endif
 
         /* j is set to the next byte to process by the previous loop. */
         for (; j < maxlen; j++) {

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/cron.cpp

@@ -51,7 +51,6 @@ void cronCommand(client *c)
         
         if (c->argc < (6 + numkeys)) {
             addReplyError(c, "Missing arguments or numkeys is too big");
-            return;
         }
     }
 

src/networking.cpp

@@ -1324,7 +1324,8 @@ void acceptOnThread(connection *conn, int flags, char *cip)
         int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT, fBootLoad] {
             connMarshalThread(conn);
             acceptCommonHandler(conn,flags,szT,ielTarget);
-            rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
+            if (!g_fTestMode && !fBootLoad)
+                rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
             zfree(szT);
         });
 
@@ -1332,7 +1333,8 @@ void acceptOnThread(connection *conn, int flags, char *cip)
             return;
         // If res != AE_OK we can still try to accept on the local thread
     }
-    rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
+    if (!g_fTestMode && !fBootLoad)
+        rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
 
     aeAcquireLock();
     acceptCommonHandler(conn,flags,cip,ielCur);
@@ -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/new.cpp

@@ -27,17 +27,14 @@ void *operator new(std::size_t size, const std::nothrow_t &) noexcept
     return zmalloc(size, MALLOC_LOCAL);
 }
 
-//need to do null checks for delete since the compiler can optimize out null checks in zfree
 void operator delete(void * p) noexcept
 {
-    if (p != nullptr)
-        zfree(p);
+    zfree(p);
 }
 
 void operator delete(void *p, std::size_t) noexcept
 {
-    if (p != nullptr)
-        zfree(p);
+    zfree(p);
 }
 
 #endif

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,31 @@ 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();
+        clock_nanosleep(CLOCK_MONOTONIC, 0, &delay, NULL);
+    }
+}
+
 void *workerThreadMain(void *parg)
 {
     int iel = (int)((int64_t)parg);
@@ -6930,6 +6962,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);