QoS for system critical events ( such as cluster heartbeats, replication streams, and slot migration)

src/ae.h

@@ -39,14 +39,18 @@
 #define AE_OK 0
 #define AE_ERR -1
 
-#define AE_NONE 0     /* No events registered. */
-#define AE_READABLE 1 /* Fire when descriptor is readable. */
-#define AE_WRITABLE 2 /* Fire when descriptor is writable. */
-#define AE_BARRIER 4  /* With WRITABLE, never fire the event if the      \
-                         READABLE event already fired in the same event  \
-                         loop iteration. Useful when you want to persist \
-                         things to disk before sending replies, and want \
-                         to do that in a group fashion. */
+#define AE_NONE 0                      /* No events registered. */
+#define AE_READABLE 1                  /* Fire when descriptor is readable. */
+#define AE_WRITABLE 2                  /* Fire when descriptor is writable. */
+#define AE_BARRIER 4                   /* With WRITABLE, never fire the event if the      \
+                                          READABLE event already fired in the same event  \
+                                          loop iteration. Useful when you want to persist \
+                                          things to disk before sending replies, and want \
+                                          to do that in a group fashion. */
+#define AE_HIGH_PRIORITY 8             /* Virtual routing mask flag: when set in aeCreateFileEvent(), \
+                                        * the event is registered on priority_apidata if available.   \
+                                        * Stripped before passing to the underlying OS multiplexer. */
+#define AE_QOS_PREEMPT_CHECK_MASK 0x03 /* Mask to check high-priority preemption once every 4 iterations */
 
 #define AE_FILE_EVENTS (1 << 0)
 #define AE_TIME_EVENTS (1 << 1)
@@ -80,6 +84,8 @@ typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientDat
 typedef void aeBeforeSleepProc(struct aeEventLoop *eventLoop);
 typedef void aeAfterSleepProc(struct aeEventLoop *eventLoop, int numevents);
 typedef int aeCustomPollProc(struct aeEventLoop *eventLoop);
+/* Callback invoked with elapsed microseconds after high-priority events are processed. */
+typedef void aeQoSStatsProc(struct aeEventLoop *eventLoop, uint64_t duration_us);
 
 /* File event structure */
 typedef struct aeFileEvent {
@@ -123,6 +129,17 @@ typedef struct aeEventLoop {
     aeCustomPollProc *custompoll;
     pthread_mutex_t poll_mutex;
     int flags;
+
+    /* High-priority event processing:
+     * Sockets registered with AE_HIGH_PRIORITY are tracked in priority_apidata.
+     * priority_fd is registered into apidata to wake the main loop when high-priority traffic arrives.
+     * priority_fired holds fired events when draining high-priority channels. */
+    aeApiState *priority_apidata;                       /* Dedicated high-priority polling state */
+    int priority_fd;                                    /* File descriptor of high-priority polling backend (-1 if disabled) */
+    aeFiredEvent *priority_fired;                       /* Fired events buffer for high-priority polling */
+    monotime priority_events_last_poll;                 /* Timestamp when high-priority events were last drained */
+    uint64_t priority_events_preempt_check_interval_us; /* Preemptive check interval in microseconds (0 = disabled) */
+    aeQoSStatsProc *priority_events_stats_callback;     /* Callback invoked with elapsed microseconds after draining high-priority events */
 } aeEventLoop;
 
 /* Prototypes */
@@ -152,4 +169,9 @@ int aeGetSetSize(aeEventLoop *eventLoop);
 int aeResizeSetSize(aeEventLoop *eventLoop, int setsize);
 void aeSetDontWait(aeEventLoop *eventLoop, int noWait);
 
+/* High-priority event loop prototypes */
+int aeActuateQoSEventLoopIfSupported(aeEventLoop *eventLoop, uint64_t qosPreemptPollIntervalUs, aeQoSStatsProc *qosStatsCallback);
+int aeProcessQoSEventsPreemptively(aeEventLoop *eventLoop);
+void aeSetQoSPreemptCheckInterval(aeEventLoop *eventLoop, uint64_t interval_us);
+
 #endif

src/ae_epoll.c

@@ -134,3 +134,7 @@ static int aeApiPoll(aeApiState *state, aeFiredEvent *fired, aeFileEvent *events
 static char *aeApiName(void) {
     return "epoll";
 }
+
+static int aeApiGetPollFd(aeApiState *state) {
+    return state ? state->epfd : -1;
+}

src/ae_evport.c

@@ -289,3 +289,8 @@ static int aeApiPoll(aeApiState *state, aeFiredEvent *fired, aeFileEvent *events
 static char *aeApiName(void) {
     return "evport";
 }
+
+static int aeApiGetPollFd(aeApiState *state) {
+    UNUSED(state);
+    return -1;
+}

src/ae_kqueue.c

@@ -180,3 +180,7 @@ static int aeApiPoll(aeApiState *state, aeFiredEvent *fired, aeFileEvent *events
 static char *aeApiName(void) {
     return "kqueue";
 }
+
+static int aeApiGetPollFd(aeApiState *state) {
+    return state ? state->kqfd : -1;
+}

src/ae_select.c

@@ -107,3 +107,8 @@ static int aeApiPoll(aeApiState *state, aeFiredEvent *fired, aeFileEvent *events
 static char *aeApiName(void) {
     return "select";
 }
+
+static int aeApiGetPollFd(aeApiState *state) {
+    UNUSED(state);
+    return -1;
+}

src/cluster.c

@@ -365,8 +365,10 @@ migrateCachedSocket *migrateGetSocket(client *c, robj *host, robj *port, long ti
         dictDelete(server.migrate_cached_sockets, dictGetKey(de));
     }
 
-    /* Create the connection */
+    /* Create the connection and tag as high-priority so key/slot migration
+     * packets are not delayed by normal tenant commands. */
     conn = connCreate(connTypeOfCluster());
+    connSetPriority(conn, true);
     if (connBlockingConnect(conn, objectGetVal(host), atoi(objectGetVal(port)), timeout) != C_OK) {
         addReplyError(c, "-IOERR error or timeout connecting to the client");
         connClose(conn);

src/cluster_legacy.c

@@ -2052,6 +2052,9 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
         }
 
         connection *conn = connCreateAccepted(connTypeOfCluster(), cfd, &require_auth);
+        /* Tag inbound cluster bus link as high-priority so cluster gossip and heartbeats
+         * are processed via QoS ahead of normal client traffic. */
+        connSetPriority(conn, true);
         /* Mark as cluster-owned before any TLS accept retries so generic
          * accept offload routing can safely avoid client assumptions. */
         connSetOwnerKind(conn, CONN_OWNER_CLUSTER_LINK);
@@ -6677,6 +6680,9 @@ static int clusterNodeCronHandleReconnect(clusterNode *node, mstime_t now, long
         (*cluster_conn_attempts)--;
         clusterLink *link = createClusterLink(node);
         link->conn = connCreate(connTypeOfCluster());
+        /* Tag outbound cluster bus link as high-priority so node reconnects, gossip ping/pong,
+         * and failure detection heartbeats operate with QoS priority. */
+        connSetPriority(link->conn, true);
         connSetPrivateData(link->conn, link);
         connSetOwnerKind(link->conn, CONN_OWNER_CLUSTER_LINK);
         if (connConnect(link->conn, node->ip, node->cport, server.bind_source_addr, 0, clusterLinkConnectHandler) ==

src/cluster_migrateslots.c

@@ -851,6 +851,12 @@ slotMigrationJob *createSlotImportJob(client *c,
     job->state = SLOT_IMPORT_WAIT_ACK;
     job->client = c;
     job->client->slot_migration_job = job;
+    if (c && c->conn) {
+        /* Upgrade connection to high priority */
+        if (connSetPriority(c->conn, true) == C_ERR) {
+            serverLog(LL_WARNING, "Failed to upgrade priority for slot migration connection %d", c->conn->fd);
+        }
+    }
 
     /* We treat slot imports like primaries. Primaries are expected to have a
      * dedicated query buffer and allocated replication data.
@@ -1379,6 +1385,8 @@ int connectSlotExportJob(slotMigrationJob *job) {
               port);
 
     job->conn = connCreate(connTypeOfReplication());
+    /* Set connection to high priority */
+    connSetPriority(job->conn, true);
     if (connConnect(job->conn, n->ip, port, server.bind_source_addr,
                     0, slotExportConnectHandler) == C_ERR) {
         return C_ERR;

src/commands.def

@@ -1470,6 +1470,7 @@ commandHistory CLIENT_KILL_History[] = {
 {"8.0.0","Replaced `master` `TYPE` with `primary`. `master` still supported for backward compatibility."},
 {"8.1.0","`ID` option accepts multiple IDs."},
 {"9.0.0","Added filters NAME, IDLE, FLAGS, LIB-NAME, LIB-VER, DB, CAPA, and IP. And negative filters NOT-ID, NOT-TYPE, NOT-ADDR, NOT-LADDR, NOT-USER, NOT-FLAGS, NOT-NAME, NOT-LIB-NAME, NOT-LIB-VER, NOT-DB, NOT-CAPA, NOT-IP."},
+{"9.2.0","Added `H` flag to indicate high priority clients."},
 };
 #endif
 
@@ -1565,6 +1566,7 @@ commandHistory CLIENT_LIST_History[] = {
 {"8.0.0","Replaced `master` `TYPE` with `primary`. `master` still supported for backward compatibility."},
 {"8.1.0","Added filters USER, ADDR, LADDR, SKIPME, and MAXAGE."},
 {"9.0.0","Added filters NAME, IDLE, FLAGS, LIB-NAME, LIB-VER, DB, CAPA, and IP. And negative filters NOT-ID, NOT-TYPE, NOT-ADDR, NOT-LADDR, NOT-USER, NOT-FLAGS, NOT-NAME, NOT-LIB-NAME, NOT-LIB-VER, NOT-DB, NOT-CAPA, NOT-IP."},
+{"9.2.0","Added `H` flag to indicate high priority clients."},
 };
 #endif
 
@@ -1912,8 +1914,8 @@ struct COMMAND_STRUCT CLIENT_Subcommands[] = {
 {MAKE_CMD("id","Returns the unique client ID of the connection.","O(1)","5.0.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_ID_History,0,CLIENT_ID_Tips,0,clientIDCommand,2,CMD_ALLOW_BUSY|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_CONNECTION|ACL_CATEGORY_SLOW,NULL,CLIENT_ID_Keyspecs,0,NULL,0)},
 {MAKE_CMD("import-source","Marks this client as an import source when the server is in import mode.","O(1)","8.1.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_IMPORT_SOURCE_History,0,CLIENT_IMPORT_SOURCE_Tips,0,clientImportSourceCommand,3,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE,ACL_CATEGORY_CONNECTION|ACL_CATEGORY_SLOW,NULL,CLIENT_IMPORT_SOURCE_Keyspecs,0,NULL,1),.args=CLIENT_IMPORT_SOURCE_Args},
 {MAKE_CMD("info","Returns information about the connection.","O(1)","6.2.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_INFO_History,0,CLIENT_INFO_Tips,1,clientInfoCommand,2,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_CONNECTION|ACL_CATEGORY_SLOW,NULL,CLIENT_INFO_Keyspecs,0,NULL,0)},
-{MAKE_CMD("kill","Terminates open connections.","O(N) where N is the number of client connections","2.4.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_KILL_History,9,CLIENT_KILL_Tips,0,clientKillCommand,-3,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_ADMIN|ACL_CATEGORY_CONNECTION|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CLIENT_KILL_Keyspecs,0,NULL,1),.args=CLIENT_KILL_Args},
-{MAKE_CMD("list","Lists open connections.","O(N) where N is the number of client connections","2.4.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_LIST_History,9,CLIENT_LIST_Tips,1,clientListCommand,-2,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_ADMIN|ACL_CATEGORY_CONNECTION|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CLIENT_LIST_Keyspecs,0,NULL,27),.args=CLIENT_LIST_Args},
+{MAKE_CMD("kill","Terminates open connections.","O(N) where N is the number of client connections","2.4.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_KILL_History,10,CLIENT_KILL_Tips,0,clientKillCommand,-3,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_ADMIN|ACL_CATEGORY_CONNECTION|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CLIENT_KILL_Keyspecs,0,NULL,1),.args=CLIENT_KILL_Args},
+{MAKE_CMD("list","Lists open connections.","O(N) where N is the number of client connections","2.4.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_LIST_History,10,CLIENT_LIST_Tips,1,clientListCommand,-2,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_ADMIN|ACL_CATEGORY_CONNECTION|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CLIENT_LIST_Keyspecs,0,NULL,27),.args=CLIENT_LIST_Args},
 {MAKE_CMD("no-evict","Sets the client eviction mode of the connection.","O(1)","7.0.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_NO_EVICT_History,0,CLIENT_NO_EVICT_Tips,0,clientNoEvictCommand,3,CMD_ALLOW_BUSY|CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_ADMIN|ACL_CATEGORY_CONNECTION|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CLIENT_NO_EVICT_Keyspecs,0,NULL,1),.args=CLIENT_NO_EVICT_Args},
 {MAKE_CMD("no-touch","Controls whether commands sent by the client affect the LRU/LFU of accessed keys.","O(1)","7.2.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_NO_TOUCH_History,0,CLIENT_NO_TOUCH_Tips,0,clientNoTouchCommand,3,CMD_ALLOW_BUSY|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE,ACL_CATEGORY_CONNECTION|ACL_CATEGORY_SLOW,NULL,CLIENT_NO_TOUCH_Keyspecs,0,NULL,1),.args=CLIENT_NO_TOUCH_Args},
 {MAKE_CMD("pause","Suspends commands processing.","O(1)","3.0.0",CMD_DOC_NONE,NULL,NULL,"connection",COMMAND_GROUP_CONNECTION,CLIENT_PAUSE_History,1,CLIENT_PAUSE_Tips,0,clientPauseCommand,-3,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_ADMIN|ACL_CATEGORY_CONNECTION|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CLIENT_PAUSE_Keyspecs,0,NULL,2),.args=CLIENT_PAUSE_Args},

src/commands/client-kill.json

@@ -43,6 +43,10 @@
             [
                 "9.0.0",
                 "Added filters NAME, IDLE, FLAGS, LIB-NAME, LIB-VER, DB, CAPA, and IP. And negative filters NOT-ID, NOT-TYPE, NOT-ADDR, NOT-LADDR, NOT-USER, NOT-FLAGS, NOT-NAME, NOT-LIB-NAME, NOT-LIB-VER, NOT-DB, NOT-CAPA, NOT-IP."
+            ],
+            [
+                "9.2.0",
+                "Added `H` flag to indicate high priority clients."
             ]
         ],
         "command_flags": [

src/commands/client-list.json

@@ -43,6 +43,10 @@
             [
                 "9.0.0",
                 "Added filters NAME, IDLE, FLAGS, LIB-NAME, LIB-VER, DB, CAPA, and IP. And negative filters NOT-ID, NOT-TYPE, NOT-ADDR, NOT-LADDR, NOT-USER, NOT-FLAGS, NOT-NAME, NOT-LIB-NAME, NOT-LIB-VER, NOT-DB, NOT-CAPA, NOT-IP."
+            ],
+            [
+                "9.2.0",
+                "Added `H` flag to indicate high priority clients."
             ]
         ],
         "command_flags": [

src/config.c

@@ -2677,6 +2677,16 @@ static int updateDefragConfiguration(const char **err) {
     return 1;
 }
 
+/* Dynamic configuration apply callback for priority-preemptive-poll-interval-us.
+ * Updates the preemption threshold on the active event loop. */
+static int updatePriorityPreemptivePollInterval(const char **err) {
+    UNUSED(err);
+    if (server.el) {
+        aeSetQoSPreemptCheckInterval(server.el, server.priority_preemptive_poll_interval_us);
+    }
+    return 1;
+}
+
 static int updateJemallocBgThread(const char **err) {
     UNUSED(err);
     set_jemalloc_bg_thread(server.jemalloc_bg_thread);
@@ -3508,6 +3518,7 @@ standardConfig static_configs[] = {
     createIntConfig("active-defrag-threshold-lower", NULL, MODIFIABLE_CONFIG, 0, 1000, server.active_defrag_threshold_lower, 10, INTEGER_CONFIG, NULL, NULL),                       /* Default: don't defrag when fragmentation is below 10% */
     createIntConfig("active-defrag-threshold-upper", NULL, MODIFIABLE_CONFIG, 0, 1000, server.active_defrag_threshold_upper, 100, INTEGER_CONFIG, NULL, updateDefragConfiguration), /* Default: maximum defrag force at 100% fragmentation */
     createIntConfig("active-defrag-cycle-us", NULL, MODIFIABLE_CONFIG, 0, 100000, server.active_defrag_cycle_us, 500, INTEGER_CONFIG, NULL, updateDefragConfiguration),
+    createIntConfig("priority-preemptive-poll-interval-us", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.priority_preemptive_poll_interval_us, 2000, INTEGER_CONFIG, NULL, updatePriorityPreemptivePollInterval),
     createIntConfig("lfu-log-factor", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, lfu_config_log_factor, 10, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("lfu-decay-time", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, lfu_config_decay_time, 1, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("replica-priority", "slave-priority", MODIFIABLE_CONFIG, 0, INT_MAX, server.replica_priority, 100, INTEGER_CONFIG, NULL, NULL),

src/connection.c

@@ -168,3 +168,51 @@ sds getListensInfoString(sds info) {
 
     return info;
 }
+/* Set connection priority. If the connection already has active events
+ * registered in the event loop, migrate them to the new priority level.
+ * Handles postponed state safely if the socket is offloaded to IO threads,
+ * and preserves AE_BARRIER ordering flags.
+ * Returns C_OK on success, or C_ERR if event migration fails. */
+int connSetPriority(connection *conn, bool is_priority) {
+    serverAssert(conn != NULL);
+    if (conn->is_priority == is_priority) return C_OK;
+
+    /* Fast path: if no socket exists yet, update priority field directly */
+    if (conn->fd == -1) {
+        conn->is_priority = is_priority;
+        return C_OK;
+    }
+
+    int mask = aeGetFileEvents(server.el, conn->fd);
+    if (mask == AE_NONE) {
+        conn->is_priority = is_priority;
+        return C_OK;
+    }
+
+    /* If socket state update is postponed by IO threads, update priority field only;
+     * connUpdateState() will register with the new priority upon IO completion. */
+    if (conn->flags & CONN_FLAG_POSTPONE_UPDATE_STATE) {
+        conn->is_priority = is_priority;
+        return C_OK;
+    }
+
+    /* Dynamic migration: active events exist on this socket */
+    bool old_priority = conn->is_priority;
+    conn->is_priority = is_priority;
+
+    /* If transport has custom state updater (e.g. TLS), delegate to it */
+    if (conn->type && conn->type->update_state) {
+        conn->type->update_state(conn);
+    } else {
+        mask = (mask & ~AE_HIGH_PRIORITY);               /* Strip off old priority flag */
+        if (conn->is_priority) mask |= AE_HIGH_PRIORITY; /* Add new priority flag */
+
+        if (aeCreateFileEvent(server.el, conn->fd, mask, conn->type->ae_handler, conn) == AE_ERR) {
+            return C_ERR;
+        }
+    }
+
+    serverLog(LL_DEBUG, "Connection fd %d priority updated from %s to %s",
+              conn->fd, old_priority ? "prioritized" : "normal", is_priority ? "prioritized" : "normal");
+    return C_OK;
+}