From 329ac612c6aaaf2c4344ad538ddd44246bfbfe83 Mon Sep 17 00:00:00 2001 From: Harry Lin Date: Thu, 29 Jan 2026 14:19:19 -0800 Subject: [PATCH 1/4] Blocked_inuse --- src/Makefile | 1 + src/blocked.c | 2 +- src/blocked_inuse.c | 379 ++++++++++++++++++++++++++++++++++++++++++++ src/blocked_inuse.h | 70 ++++++++ src/networking.c | 5 + src/server.c | 51 ++++++ src/server.h | 3 + src/timeout.c | 2 + 8 files changed, 512 insertions(+), 1 deletion(-) create mode 100644 src/blocked_inuse.c create mode 100644 src/blocked_inuse.h diff --git a/src/Makefile b/src/Makefile index a0fe81f5b0c..7c80d33c3c1 100644 --- a/src/Makefile +++ b/src/Makefile @@ -460,6 +460,7 @@ ENGINE_SERVER_OBJ = \ bio.o \ bitops.o \ blocked.o \ + blocked_inuse.o \ call_reply.o \ childinfo.o \ cluster.o \ diff --git a/src/blocked.c b/src/blocked.c index 83437dcf144..c46e7d43a63 100644 --- a/src/blocked.c +++ b/src/blocked.c @@ -177,7 +177,7 @@ void processUnblockedClients(void) { * is blocked again. Actually processInputBuffer() checks that the * client is not blocked before to proceed, but things may change and * the code is conceptually more correct this way. */ - if (!c->flag.blocked) { + if (!c->flag.blocked && !blockInuse_isBlockedClient(c) && !blockInuse_isUnblockedClient(c)) { /* If we have a queued command, execute it now. */ if (processPendingCommandAndInputBuffer(c) == C_ERR) { continue; diff --git a/src/blocked_inuse.c b/src/blocked_inuse.c new file mode 100644 index 00000000000..a046c287888 --- /dev/null +++ b/src/blocked_inuse.c @@ -0,0 +1,379 @@ +#include "server.h" +#include "hashtable.h" +#include "blocked_inuse.h" + +/* External functions from server.c */ +extern uint64_t dictEncObjHash(const void *key); +extern int hashtableEncObjKeyCompare(const void *key1, const void *key2); +extern uint64_t hashtableClientHash(const void *key); +extern int hashtableClientKeyCompare(const void *key1, const void *key2); + +static hashtable *client_to_keys; // client memory address -> blockInuse_clientMetadata (contains a list of keys) +static hashtable *key_to_clients; /* Map(key)->list(blocked clients). */ +static list *unblocked_clients; /* list of clients which are unblocked. We need to resume the blocked command and then process the unread querybuf before we remove it.*/ +uint32_t blocked_clients_on_keys; /* Num of clients blocked on keys now. */ +uint32_t total_clients_blocked_on_keys_lifetime; /* Num of clients blocked on keys lifetime. */ +uint32_t total_clients_unblocked_on_keys_lifetime; /* Num of clients unblocked on keys lifetime. */ +uint32_t total_clients_resumed_lifetime; /* Num of clients resumed lifetime. */ + +typedef struct blockInuse_clientMetadata { + robj **keys; /* Keys this client is blocked on. */ + int n_keys; /* number of keys this client is blocked on. */ + long long blocked_at; /* what time this client is blocked in milli second. It uses value from server.mstime. */ +} blockInuse_clientMetadata; + +/* ----------------------------- client_to_keys Hashtable Util ------------------------- */ +typedef struct { + client *c; + blockInuse_clientMetadata metadata; +} clientDataEntry; + +/* Hashtable callbacks for client -> blockInuse_clientMetadata */ +static const void *clientDataEntryGetKey(const void *entry) { + return ((clientDataEntry *)entry)->c; +} + +static void clientDataEntryDestructor(void *entry) { + clientDataEntry *e = entry; + if (e->metadata.keys) { + //Refcount for all the keys should be decreased before calling this function. Hence n_keys should be 0. + serverAssert(e->metadata.keys == 0); + for (int i = 0; i < e->metadata.n_keys; i++) { + decrRefCount(e->metadata.keys[i]); + } + zfree(e->metadata.keys); + } + zfree(entry); +} + +static hashtableType clientDataHashtableType = { + .entryGetKey = clientDataEntryGetKey, + .hashFunction = hashtableClientHash, + .keyCompare = hashtableClientKeyCompare, + .entryDestructor = clientDataEntryDestructor, +}; + +/* Utility functions for client_to_keys hashtable */ +static blockInuse_clientMetadata *addOrFindClientMetadata(client *c) { + clientDataEntry *entry; + if (hashtableFind(client_to_keys, c, (void **)&entry)) { + return &entry->metadata; + } + + entry = zcalloc(sizeof(clientDataEntry)); + entry->c = c; + hashtableAdd(client_to_keys, entry); + return &entry->metadata; +} + +static blockInuse_clientMetadata *getClientMetadata(client *c) { + clientDataEntry *entry; + if (hashtableFind(client_to_keys, c, (void **)&entry)) { + return &entry->metadata; + } + return NULL; +} + +static void removeClientMetadata(client *c) { + hashtablePop(client_to_keys, c, NULL); +} + + +/* ----------------------------- keyToClientsEntry Hashtable Util ------------------------- */ +/* Entry type for key_to_clients: robj key -> list of clients */ +typedef struct { + robj *key; + list *clients; +} keyToClientsEntry; + +static const void *keyToClientsGetKey(const void *entry) { + return ((keyToClientsEntry *)entry)->key; +} + +static void keyToClientsDestructor(void *entry) { + keyToClientsEntry *e = entry; + decrRefCount(e->key); + listRelease(e->clients); + zfree(entry); +} + +static hashtableType keyToClientsHashtableType = { + .entryGetKey = keyToClientsGetKey, + .hashFunction = dictEncObjHash, + .keyCompare = hashtableEncObjKeyCompare, + .entryDestructor = keyToClientsDestructor, +}; + +static list *addOrFindBlockedClientsListsByKey(robj *key) { + keyToClientsEntry *entry; + if (hashtableFind(key_to_clients, key, (void **)&entry)) { + return entry->clients; + } + + entry = zcalloc(sizeof(keyToClientsEntry)); + entry->key = key; + incrRefCount(key); + entry->clients = listCreate(); + hashtableAdd(key_to_clients, entry); + return entry->clients; +} + +static list *getBlockedClientsListsByKey(robj *key) { + keyToClientsEntry *entry; + if (hashtableFind(key_to_clients, key, (void **)&entry)) { + return entry->clients; + } + return NULL; +} + +static void removeBlockedClientsListsByKey(robj *key) { + hashtablePop(key_to_clients, key, NULL); +} + +/* ----------------------------- util ------------------------- */ +static void markClientBlocked(client *c) { + c->flag.blockInuse_blocked = 1; + c->flag.pending_command = 1; // Harrt TODO do we need this, and why? +} + +static void markClientUnblocked(client *c) { + c->flag.blockInuse_blocked = 0; + c->flag.blockInuse_unblocked = 1; +} + +// Init the client Metadata and insert it into our global hash table, key is client, and value is metadata +static blockInuse_clientMetadata *initClientMetadata(client *c, int nKeys) { + serverAssert(!getClientMetadata(c)); // this client must not be in our global table + serverAssert(nKeys >= 0); // non negative check + blockInuse_clientMetadata *metadata = addOrFindClientMetadata(c); // add metada into the hashtable + metadata->n_keys = 0; + metadata->keys = NULL; + metadata->blocked_at = server.mstime; + if (nKeys > 0) { + metadata->keys = zmalloc(sizeof(robj *) * nKeys); + } + return metadata; +} + +// Remove this client totally from every lists in the key_to_clients table. +static void unlinkBlockedClientOnKeys(client *c) { + blockInuse_clientMetadata *metadata = getClientMetadata(c); + if (!metadata) return; + for (int i = 0; i < metadata->n_keys; ++i) { + robj *key = metadata->keys[i]; + + list *clientList = getBlockedClientsListsByKey(key); + serverAssert(clientList != NULL); + listDelNode(clientList, listSearchKey(clientList, c)); + + if (listLength(clientList) == 0) removeBlockedClientsListsByKey(key); + decrRefCount(key); // Harry TODO: double check where to increase and where to decrease. + metadata->keys[i] = NULL; + } + metadata->n_keys = 0; + blocked_clients_on_keys--; + total_clients_unblocked_on_keys_lifetime++; + total_clients_resumed_lifetime++; // Harry TODO why ?? +} + +// Remove a key from a client entry in client_to_keys table +static blockInuse_clientMetadata *removeBlockingKeyFromClient(client *c, robj *key) { + blockInuse_clientMetadata *metadata = getClientMetadata(c); + if (metadata == NULL) return NULL; + sds key_sds = objectGetKey(key); + for (int i = 0; i < metadata->n_keys; ++i) { + sds curr_key = objectGetKey(metadata->keys[i]); + if (sdscmp(curr_key, key_sds) == 0) { + decrRefCount(metadata->keys[i]); + metadata->keys[i] = metadata->keys[metadata->n_keys - 1]; + metadata->keys[metadata->n_keys - 1] = NULL; + metadata->n_keys--; + return metadata; + } + } + // we expect to find a key + serverAssert(false); +} + +/* Process all the commands for an unblocked client. First we read the blocked command which is already parsed by calling processCommand. + * Then we process the commands present in querybuf by calling processInputBuffer. + */ +static void processBlockedCommand(client *c) { + if (c->flag.close_asap) return; + c->flag.pending_command = 0; + int retval = processCommandAndResetClient(c); + if (retval != C_OK || blockInuse_isBlockedClient(c)) { + return; + } + //process the pending commands in the buffer. + if (processInputBuffer(c) == C_OK && !c->flag.close_asap) { + beforeNextClient(c); + } +} +/* ----------------------------- API implementation ------------------------- */ + +/* Initialize global client_to_keys hashtable. Call once at server startup. */ +/* Initializes the blockInuse data structures needed for DB. Called at server startup per db. */ +// Harry check Done +void blockInuse_init(void) { + client_to_keys = hashtableCreate(&clientDataHashtableType); + key_to_clients = hashtableCreate(&keyToClientsHashtableType); + unblocked_clients = listCreate(); + blocked_clients_on_keys = 0; + total_clients_blocked_on_keys_lifetime = 0; + total_clients_unblocked_on_keys_lifetime = 0; + total_clients_resumed_lifetime = 0; +} + +/* Clean up the blockInuse data structures for the database if possible. + * Returns false if there are any blocked or unblocked clients (no cleanup performed). + * Returns true if cleanup succeeded or was already done. + * Note: This will not free the struct itself but cleans up the internal data. */ +// Harry check Done +bool blockInuse_cleanDbBlockingInfo(void) { + if (blocked_clients_on_keys > 0 || (listLength(unblocked_clients) > 0)) return false; + hashtableRelease(key_to_clients); + listRelease(unblocked_clients); + key_to_clients = NULL; + unblocked_clients = NULL; + blocked_clients_on_keys = 0; + return true; +} + +// Harry check Done +int blockInuse_getNumberOfBlockedClients(void) { + return blocked_clients_on_keys; +} + +long blockInuse_getNumberOfUnblockedClients(void) { + return listLength(unblocked_clients); +} + +// Harry check Done +int blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]) { + // some checks + serverAssert(!(blockInuse_isBlockedClient(c) || blockInuse_isUnblockedClient(c))); + if (nKeys == 0) return C_ERR; + if (c->flag.replica) return C_ERR; // Maybe remove this? + for (int i = 0; i < nKeys; ++i) { + if (keys[i]->type != OBJ_STRING) return C_ERR; + } + + // add into the global table + blockInuse_clientMetadata *metadata = initClientMetadata(c, nKeys); // this will add into the global table and assign memory, but the keys in metadata is still empty + markClientBlocked(c); + for (int i = 0; i < nKeys; ++i) { + // This loop is just for book keeping, we want to do 2 things in the loop: + // 1. adding the key into the client metadata (global table), such that c -> list of keys (append key to tail) + // 2. add the entry into the blocking info, such that key -> list of clients (append c to tail) + + list *blockedClientsList = addOrFindBlockedClientsListsByKey(keys[i]); // this will add entry of key -> [client 1, client 2 ...], incrRefCount in included. + + // If the last client blocked on this key is not c, then we add c onto the list. + // Otherwise this is a duplicated key and we should ignore it. + listNode *last_client = listLast(blockedClientsList); + + // this if check is for deduplica of keys + if (last_client == NULL || last_client->value != c) { + // 2. add the client into the blocking info at tail, either create a new entry or add to tail + listAddNodeTail(blockedClientsList, c); + + // 1. add the key into the client metadata, so it would be client -> [key1, key2, key3] + incrRefCount(keys[i]); + metadata->keys[metadata->n_keys] = keys[i]; + metadata->n_keys++; + } + } + + if (c->conn) { + // Delete the readable event from the event loop for the blocked client. + connSetReadHandler(c->conn, NULL); + } + blocked_clients_on_keys++; + total_clients_blocked_on_keys_lifetime++; + return C_OK; +} + +/* Unblock given key. A client will be unblocked, if it has no more dependency on any key and will be put into unblocked_clients list. */ +// Harry check Done +void blockInuse_unblockClientsOnKey(robj *key) { + list *blockedClientsList = getBlockedClientsListsByKey(key); + if (blockedClientsList == NULL) return; + serverAssert(listLength(blockedClientsList) > 0); + while (listLength(blockedClientsList) > 0) { + listNode *ln = listFirst(blockedClientsList); + client *c = listNodeValue(ln); + listDelNode(blockedClientsList, ln); + blockInuse_clientMetadata *metadata = removeBlockingKeyFromClient(c, key); + if (metadata->n_keys == 0) { + // time to remove this entry in our global table + markClientUnblocked(c); + removeClientMetadata(c); + listAddNodeTail(unblocked_clients, c); + blocked_clients_on_keys--; + total_clients_blocked_on_keys_lifetime++; + } + } + removeBlockedClientsListsByKey(key); +} + +// Harry check Done +void blockInuse_unblockClientsOnAllKeys(void) { + hashtableIterator iter; + hashtableInitIterator(&iter, key_to_clients, HASHTABLE_ITER_SAFE); + void *entry; + while (hashtableNext(&iter, &entry)) { + keyToClientsEntry *e = entry; + robj *key = e->key; + incrRefCount(key); + blockInuse_unblockClientsOnKey(key); + decrRefCount(key); + } + hashtableCleanupIterator(&iter); +} + +// Harry check Done +void blockInuse_processServerBlockedClients(void) { + if(!unblocked_clients) return; + + while(listLength(unblocked_clients) > 0) { + // we need to check it every time, so that if one of the unblocked + // clients executed pause command, then we stop processing further. + if (isPausedActionsWithUpdate(PAUSE_ACTIONS_CLIENT_ALL_SET)) return; + listNode *ln = listFirst(unblocked_clients); + serverAssert(ln != NULL); + client *c = listNodeValue(ln); + listDelNode(unblocked_clients, ln); + c->flag.blockInuse_unblocked = 0; + total_clients_resumed_lifetime++; + // make the fd readble again. This should succeed as we are not adding a + // new client. If it fails because epoll_ctl failed then freeClient. + // We avoid setting the read handler for fake client that does not have a connection. + if (c->conn && connSetReadHandler(c->conn, readQueryFromClient) == C_ERR) { + freeClient(c); + return; + } + processBlockedCommand(c); + } +} + +// Harry check Done +// remove a client from everywhere +void blockInuse_unlinkClient(client *c) { + serverAssert(blockInuse_isBlockedClient(c) || blockInuse_isUnblockedClient(c)); + if (blockInuse_isBlockedClient(c)) { + blockInuse_clientMetadata *metadata = getClientMetadata(c); + if (metadata == NULL) return; // return immediately if the client was not blocked on any keys. + unlinkBlockedClientOnKeys(c); + c->flag.blockInuse_blocked = 0; + removeClientMetadata(c); // remove the global hashtable entry + } + + if (blockInuse_isUnblockedClient(c)) { + listNode *ln = listSearchKey(unblocked_clients, c); + serverAssert(ln != NULL); + listDelNode(unblocked_clients, ln); + c->flag.blockInuse_unblocked = 0; + total_clients_resumed_lifetime++; + } +} diff --git a/src/blocked_inuse.h b/src/blocked_inuse.h new file mode 100644 index 00000000000..fdb8f26aa96 --- /dev/null +++ b/src/blocked_inuse.h @@ -0,0 +1,70 @@ +/* +Harry TODO: +1. write a brief introduction here +2. remove unblocked client once done +3. do we need the oldest? +*/ + +#ifndef BLOCKED_INUSE_H__ +#define BLOCKED_INUSE_H__ + +#include "hashtable.h" +#include "adlist.h" + +struct robj; //defined in server.h +struct serverCommand; //defined in server.h +struct client; //defined in server.h + +/* Check if client is blocked/unblocked by blockInuse */ +#define blockInuse_isBlockedClient(c) ((c)->flag.blockInuse_blocked) +#define blockInuse_isUnblockedClient(c) ((c)->flag.blockInuse_unblocked) + +// Harry Check: do I need them in here, since they are totally private now. +typedef struct blockInuse_clientMetadata blockInuse_clientMetadata; +typedef struct blockInuse_blockingInfo blockInuse_blockingInfo; + +/* Initialize global blockInuse structures. Call once at server startup. */ +void blockInuse_init(void); + +/* If no clients are blocked and all the previously blocked clients have processed the blocked command, + then this will free the data structures and return true. Otherwise, this will return false, and we + still have few clients to process and cleanup failed. */ +bool blockInuse_cleanDbBlockingInfo(void); + +/* Returns the total count of currently blocked clients by blockInuse */ +int blockInuse_getNumberOfBlockedClients(void); +long blockInuse_getNumberOfUnblockedClients(void); + +/* + * Block given client on set of keys. Duplicated keys are handled. + * To avoid the extra copy, we keep reference to the passed keys. So passed variable keys, should be heap allocated. + * API asserts that the client do not already have a blocked/unblocked flag set. + * Return Value: + * C_ERR: if + * a. Any passed key is not sds. + * b. nKeys = 0. + * c. Client is slave client. + * Otherwise, it blocks the client and returns C_OK. + * */ +// Blocks a client on a set of keys. +// Then client will remain blocked until all keys are unblocked. +int blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]); + +/* Unblock given key. A client will be unblocked if it has no more dependency on any key and will be + * put into unblocked_clients list. Clients from this list are processed during processUnblockedClients. + */ +void blockInuse_unblockClientsOnKey(robj *key); + +/* Unblock all clients on all keys */ +void blockInuse_unblockClientsOnAllKeys(void); + +/* If clientBlocking is enabled, this function is called in beforeSleep each time, to resume clients which were previously blocked. */ +void blockInuse_processServerBlockedClients(void); + +/* + * This API is to force unlinking of a blocked client. Typically required when we want to free the client while its blocked (e.g. memory pressure). + * This will clean up the current command arguments and detach all the references in blocking structures. + */ +void blockInuse_unlinkClient(client *c); + +#endif diff --git a/src/networking.c b/src/networking.c index eb8dcd29806..3c1dde147b8 100644 --- a/src/networking.c +++ b/src/networking.c @@ -1900,6 +1900,8 @@ void unlinkClient(client *c) { /* If this is marked as current client unset it. */ if (c->conn && server.current_client == c) server.current_client = NULL; + blockInuse_unlinkClient(c); + /* Certain operations must be done only if the client has an active connection. * If the client was already unlinked or if it's a "fake client" the * conn is already set to NULL. */ @@ -1983,6 +1985,9 @@ void unlinkClient(client *c) { /* Clear the tracking status. */ if (c->flag.tracking) disableTracking(c); + + // We should never have a client here which is in blockInuse unblocked or blocked state. + serverAssert(!(blockInuse_isBlockedClient(c) || blockInuse_isUnblockedClient(c))); } /* Clear the client state to resemble a newly connected client. */ diff --git a/src/server.c b/src/server.c index d2ab7ce115c..87807b7c968 100644 --- a/src/server.c +++ b/src/server.c @@ -75,6 +75,7 @@ #include #include #include +#include #ifdef __linux__ #include @@ -1153,6 +1154,42 @@ void getExpensiveClientsInfo(size_t *in_usage, size_t *out_usage) { *out_usage = o; } +// return 1 if client was terminated, 0 if still alive. +static int clientsCronCheckBlockInuseClients(client *c) { + // Check for clients with no read/write handlers (blocked clients) + // which have been closed from the remote side. + if (c->conn) { + // It's a normal client connection (not a fake client) ... + if (c->conn->type == connectionTypeTcp() || c->conn->type == connectionTypeTls()) { + // ... and it's based on a TCP socket ... + if (aeGetFileEvents(server.el, c->conn->fd) == AE_NONE) { + // ... and neither read nor write handler is installed ... + // Determine if the connection has been closed, from the far end, by + // checking the TCP state information. + struct tcp_info info; + socklen_t infolen = sizeof(info); + // Query the kernal for TCP socket state info + // since no event handler exists, we must manally check if the connection is dead. + if (getsockopt(c->conn->fd, IPPROTO_TCP, TCP_INFO, &info, &infolen) == 0) { + // check TCP state + if (info.tcpi_state == TCP_CLOSE_WAIT || info.tcpi_state == TCP_CLOSE) { + // TCP_CLOSE_WAIT: remote side closed, local side hasn't closed yet + // TCP_CLOSE: connection fully closed. + if (server.verbosity <= LL_VERBOSE) { + sds info = catClientInfoString(sdsempty(), c, server.hide_user_data_from_log); + serverLog(LL_VERBOSE, "Client closed connection while blocked %s", info); + sdsfree(info); + } + freeClientAsync(c); + return 1; // client has been closed + } + } + } + } + } + return 0; // client has not been terminated +} + /* This function is called by clientsTimeProc() and is used in order to perform * operations on clients that are important to perform constantly. For instance * we use this function in order to disconnect clients after a timeout, including @@ -1207,6 +1244,7 @@ static void clientsCron(int clients_this_cycle) { if (clientsCronResizeQueryBuffer(c)) continue; if (clientsCronResizeOutputBuffer(c, now)) continue; if (clientsCronTrackExpensiveClients(c, curr_peak_mem_usage_slot)) continue; + if (clientsCronCheckBlockInuseClients(c)) continue; /* Iterating all the clients in getMemoryOverheadData() is too slow and * in turn would make the INFO command too slow. So we perform this @@ -1943,6 +1981,10 @@ void beforeSleep(struct aeEventLoop *eventLoop) { /* Close clients that need to be closed asynchronous */ freeClientsInAsyncFreeQueue(); + if (blockInuse_getNumberOfUnblockedClients() > 0) { + blockInuse_processServerBlockedClients(); + } + /* Incrementally trim replication backlog, 10 times the normal speed is * to free replication backlog as much as possible. */ if (server.repl_backlog) incrementalTrimReplicationBacklog(10 * REPL_BACKLOG_TRIM_BLOCKS_PER_CALL); @@ -2912,6 +2954,9 @@ void initServer(void) { server.debug_client_enforce_reply_list = 0; resetReplicationBuffer(); + /* Init blockInuse */ + blockInuse_init(); + /* Make sure the locale is set on startup based on the config file. */ if (setlocale(LC_COLLATE, server.locale_collate) == NULL) { serverLog(LL_WARNING, "Failed to configure LOCALE for invalid locale name."); @@ -4180,6 +4225,9 @@ void unprepareCommand(client *c) { * other operations can be performed by the caller. Otherwise * if C_ERR is returned the client was destroyed (i.e. after QUIT). */ int processCommand(client *c) { + + serverAssert(!(blockInuse_isBlockedClient(c) || blockInuse_isUnblockedClient(c))); + if (!scriptIsTimedout()) { /* Both EXEC and scripts call call() directly so there should be * no way in_exec or scriptIsRunning() is 1. @@ -4828,6 +4876,9 @@ int finishShutdown(void) { /* Close the listening sockets. Apparently this allows faster restarts. */ closeListeningSockets(1); + /* Cleanup blockInuse data structures */ + blockInuse_cleanDbBlockingInfo(); + moduleUnloadAllModules(); serverLog(LL_WARNING, "%s is now ready to exit, bye bye...", server.sentinel_mode ? "Sentinel" : "Valkey"); diff --git a/src/server.h b/src/server.h index 51a491ea4d5..f9c259a2ca6 100644 --- a/src/server.h +++ b/src/server.h @@ -83,6 +83,7 @@ #include "trace/trace.h" #include "entry.h" #include "lrulfu.h" +#include "blocked_inuse.h" /* * Sanity check: we require large-file support. If include order caused @@ -1151,6 +1152,8 @@ typedef struct ClientFlags { uint64_t dirty_cas : 1; /* Watched keys modified. EXEC will fail. */ uint64_t close_after_reply : 1; /* Close after writing entire reply. */ uint64_t unblocked : 1; /* This client was unblocked and is stored in server.unblocked_clients */ + uint64_t blockInuse_blocked : 1; /* This client is blocked by blockInuse */ + uint64_t blockInuse_unblocked : 1; /* This client is unblocked by blockInuse */ uint64_t script : 1; /* This is a non connected client used by Lua */ uint64_t asking : 1; /* Client issued the ASKING command */ uint64_t close_asap : 1; /* Close this client ASAP */ diff --git a/src/timeout.c b/src/timeout.c index 8ff0608ae0a..fb6c1ba8fe3 100644 --- a/src/timeout.c +++ b/src/timeout.c @@ -59,6 +59,8 @@ int clientsCronHandleTimeout(client *c, mstime_t now_ms) { !mustObeyClient(c) && /* No timeout for primaries and AOF */ !c->flag.blocked && /* No timeout for BLPOP */ !c->flag.pubsub && /* No timeout for Pub/Sub clients */ + !c->flag.blockInuse_blocked && /* No timeout for BlockInuse client */ + !c->flag.blockInuse_unblocked && /* Client is unblocked, but we haven't yet processed the blocking command and input buffer, no timeout */ (now - c->last_interaction > server.maxidletime)) { serverLog(LL_VERBOSE, "Closing idle client"); freeClient(c); From 28aa8c8cc087b76bb299b83fa84097d917428947 Mon Sep 17 00:00:00 2001 From: Harry Lin Date: Tue, 3 Feb 2026 21:08:52 -0800 Subject: [PATCH 2/4] Combine the unblocked client lists --- cmake/Modules/SourceFiles.cmake | 1 + src/blocked.c | 18 +++- src/blocked_inuse.c | 145 ++++++++++++-------------------- src/blocked_inuse.h | 49 +++++++---- src/networking.c | 4 +- src/server.c | 8 +- 6 files changed, 105 insertions(+), 120 deletions(-) diff --git a/cmake/Modules/SourceFiles.cmake b/cmake/Modules/SourceFiles.cmake index 6081c2d2e45..8cd97cd5b01 100644 --- a/cmake/Modules/SourceFiles.cmake +++ b/cmake/Modules/SourceFiles.cmake @@ -64,6 +64,7 @@ set(VALKEY_SERVER_SRCS ${CMAKE_SOURCE_DIR}/src/notify.c ${CMAKE_SOURCE_DIR}/src/setproctitle.c ${CMAKE_SOURCE_DIR}/src/blocked.c + ${CMAKE_SOURCE_DIR}/src/server_blocking.c ${CMAKE_SOURCE_DIR}/src/hyperloglog.c ${CMAKE_SOURCE_DIR}/src/latency.c ${CMAKE_SOURCE_DIR}/src/sparkline.c diff --git a/src/blocked.c b/src/blocked.c index c46e7d43a63..3d9b61f6522 100644 --- a/src/blocked.c +++ b/src/blocked.c @@ -160,6 +160,9 @@ void processUnblockedClients(void) { client *c; while (listLength(server.unblocked_clients)) { + // we need to check it every time, so that if one of the unblocked + // clients executed pause command, then we stop processing further. + if (isPausedActionsWithUpdate(PAUSE_ACTIONS_CLIENT_ALL_SET)) return; ln = listFirst(server.unblocked_clients); serverAssert(ln != NULL); c = ln->value; @@ -173,17 +176,28 @@ void processUnblockedClients(void) { continue; } + if (blockInuse_isBlockedClient(c)) { + c->flag.blockInuse_unblocked = 0; + // make the fd readble again. This should succeed as we are not adding a + // new client. If it fails because epoll_ctl failed then freeClient. + // We avoid setting the read handler for fake client that does not have a connection. + if (c->conn && connSetReadHandler(c->conn, readQueryFromClient) == C_ERR) { + freeClient(c); + return; + } + if (c->flag.close_asap) return; // maybe move out of the if? + } /* Process remaining data in the input buffer, unless the client * is blocked again. Actually processInputBuffer() checks that the * client is not blocked before to proceed, but things may change and * the code is conceptually more correct this way. */ - if (!c->flag.blocked && !blockInuse_isBlockedClient(c) && !blockInuse_isUnblockedClient(c)) { + if (!c->flag.blocked && !blockInuse_isBlockedClient(c)) { /* If we have a queued command, execute it now. */ if (processPendingCommandAndInputBuffer(c) == C_ERR) { continue; } } - beforeNextClient(c); + if (c) beforeNextClient(c); } } diff --git a/src/blocked_inuse.c b/src/blocked_inuse.c index a046c287888..40aa1c68d9d 100644 --- a/src/blocked_inuse.c +++ b/src/blocked_inuse.c @@ -1,28 +1,26 @@ #include "server.h" -#include "hashtable.h" #include "blocked_inuse.h" -/* External functions from server.c */ +/* External hashtable functions from server.c */ extern uint64_t dictEncObjHash(const void *key); extern int hashtableEncObjKeyCompare(const void *key1, const void *key2); extern uint64_t hashtableClientHash(const void *key); extern int hashtableClientKeyCompare(const void *key1, const void *key2); -static hashtable *client_to_keys; // client memory address -> blockInuse_clientMetadata (contains a list of keys) -static hashtable *key_to_clients; /* Map(key)->list(blocked clients). */ -static list *unblocked_clients; /* list of clients which are unblocked. We need to resume the blocked command and then process the unread querybuf before we remove it.*/ -uint32_t blocked_clients_on_keys; /* Num of clients blocked on keys now. */ -uint32_t total_clients_blocked_on_keys_lifetime; /* Num of clients blocked on keys lifetime. */ -uint32_t total_clients_unblocked_on_keys_lifetime; /* Num of clients unblocked on keys lifetime. */ -uint32_t total_clients_resumed_lifetime; /* Num of clients resumed lifetime. */ +// Internal blockInuse data structure +static hashtable *client_to_keys; /* Map(client memory address) -> blockInuse_clientMetadata (contains a list of keys). */ +static hashtable *key_to_clients; /* Map(key)->list(blocked clients). */ +uint32_t blocked_clients_on_keys; /* Num of clients blocked on keys now. */ +uint32_t blockInuse_total_clients_blocked_on_keys_lifetime; /* Num of clients blocked on keys lifetime. */ +uint32_t blockInuse_total_clients_unblocked_on_keys_lifetime; /* Num of clients unblocked on keys lifetime. */ typedef struct blockInuse_clientMetadata { robj **keys; /* Keys this client is blocked on. */ - int n_keys; /* number of keys this client is blocked on. */ - long long blocked_at; /* what time this client is blocked in milli second. It uses value from server.mstime. */ + int n_keys; /* Number of keys this client is blocked on. */ + long long blocked_at; /* What time this client is blocked in milli second. It uses value from server.mstime. */ } blockInuse_clientMetadata; -/* ----------------------------- client_to_keys Hashtable Util ------------------------- */ +/* ----------------------------- client_to_keys Hashtable util ------------------------- */ typedef struct { client *c; blockInuse_clientMetadata metadata; @@ -79,7 +77,7 @@ static void removeClientMetadata(client *c) { } -/* ----------------------------- keyToClientsEntry Hashtable Util ------------------------- */ +/* ----------------------------- key_to_clients Hashtable Util ------------------------- */ /* Entry type for key_to_clients: robj key -> list of clients */ typedef struct { robj *key; @@ -104,6 +102,7 @@ static hashtableType keyToClientsHashtableType = { .entryDestructor = keyToClientsDestructor, }; +/* Utility functions for key_to_clients hashtable */ static list *addOrFindBlockedClientsListsByKey(robj *key) { keyToClientsEntry *entry; if (hashtableFind(key_to_clients, key, (void **)&entry)) { @@ -136,12 +135,7 @@ static void markClientBlocked(client *c) { c->flag.pending_command = 1; // Harrt TODO do we need this, and why? } -static void markClientUnblocked(client *c) { - c->flag.blockInuse_blocked = 0; - c->flag.blockInuse_unblocked = 1; -} - -// Init the client Metadata and insert it into our global hash table, key is client, and value is metadata +// Init the client Metadata and insert it into client_to_keys hash table. static blockInuse_clientMetadata *initClientMetadata(client *c, int nKeys) { serverAssert(!getClientMetadata(c)); // this client must not be in our global table serverAssert(nKeys >= 0); // non negative check @@ -172,8 +166,7 @@ static void unlinkBlockedClientOnKeys(client *c) { } metadata->n_keys = 0; blocked_clients_on_keys--; - total_clients_unblocked_on_keys_lifetime++; - total_clients_resumed_lifetime++; // Harry TODO why ?? + blockInuse_total_clients_unblocked_on_keys_lifetime++; } // Remove a key from a client entry in client_to_keys table @@ -195,21 +188,6 @@ static blockInuse_clientMetadata *removeBlockingKeyFromClient(client *c, robj *k serverAssert(false); } -/* Process all the commands for an unblocked client. First we read the blocked command which is already parsed by calling processCommand. - * Then we process the commands present in querybuf by calling processInputBuffer. - */ -static void processBlockedCommand(client *c) { - if (c->flag.close_asap) return; - c->flag.pending_command = 0; - int retval = processCommandAndResetClient(c); - if (retval != C_OK || blockInuse_isBlockedClient(c)) { - return; - } - //process the pending commands in the buffer. - if (processInputBuffer(c) == C_OK && !c->flag.close_asap) { - beforeNextClient(c); - } -} /* ----------------------------- API implementation ------------------------- */ /* Initialize global client_to_keys hashtable. Call once at server startup. */ @@ -218,24 +196,19 @@ static void processBlockedCommand(client *c) { void blockInuse_init(void) { client_to_keys = hashtableCreate(&clientDataHashtableType); key_to_clients = hashtableCreate(&keyToClientsHashtableType); - unblocked_clients = listCreate(); blocked_clients_on_keys = 0; - total_clients_blocked_on_keys_lifetime = 0; - total_clients_unblocked_on_keys_lifetime = 0; - total_clients_resumed_lifetime = 0; + blockInuse_total_clients_blocked_on_keys_lifetime = 0; + blockInuse_total_clients_unblocked_on_keys_lifetime = 0; } /* Clean up the blockInuse data structures for the database if possible. * Returns false if there are any blocked or unblocked clients (no cleanup performed). * Returns true if cleanup succeeded or was already done. * Note: This will not free the struct itself but cleans up the internal data. */ -// Harry check Done -bool blockInuse_cleanDbBlockingInfo(void) { - if (blocked_clients_on_keys > 0 || (listLength(unblocked_clients) > 0)) return false; +void blockInuse_release(void) { + serverAssert(blocked_clients_on_keys == 0); hashtableRelease(key_to_clients); - listRelease(unblocked_clients); key_to_clients = NULL; - unblocked_clients = NULL; blocked_clients_on_keys = 0; return true; } @@ -245,14 +218,10 @@ int blockInuse_getNumberOfBlockedClients(void) { return blocked_clients_on_keys; } -long blockInuse_getNumberOfUnblockedClients(void) { - return listLength(unblocked_clients); -} - // Harry check Done int blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]) { // some checks - serverAssert(!(blockInuse_isBlockedClient(c) || blockInuse_isUnblockedClient(c))); + serverAssert(!(blockInuse_isBlockedClient(c) || (c)->flag.unblocked)); if (nKeys == 0) return C_ERR; if (c->flag.replica) return C_ERR; // Maybe remove this? for (int i = 0; i < nKeys; ++i) { @@ -290,7 +259,7 @@ int blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]) { connSetReadHandler(c->conn, NULL); } blocked_clients_on_keys++; - total_clients_blocked_on_keys_lifetime++; + blockInuse_total_clients_blocked_on_keys_lifetime++; return C_OK; } @@ -304,16 +273,22 @@ void blockInuse_unblockClientsOnKey(robj *key) { listNode *ln = listFirst(blockedClientsList); client *c = listNodeValue(ln); listDelNode(blockedClientsList, ln); + // remove a key for a specific client in client_to_keys blockInuse_clientMetadata *metadata = removeBlockingKeyFromClient(c, key); if (metadata->n_keys == 0) { // time to remove this entry in our global table - markClientUnblocked(c); + serverAssert(c->flag.unblocked == 0); + if (!c->flag.unblocked) { + c->flag.unblocked = 1; + listAddNodeTail(server.unblocked_clients, c); + } removeClientMetadata(c); - listAddNodeTail(unblocked_clients, c); blocked_clients_on_keys--; - total_clients_blocked_on_keys_lifetime++; + blockInuse_total_clients_blocked_on_keys_lifetime++; } } + + // remove from key_to_clients removeBlockedClientsListsByKey(key); } @@ -332,48 +307,34 @@ void blockInuse_unblockClientsOnAllKeys(void) { hashtableCleanupIterator(&iter); } -// Harry check Done -void blockInuse_processServerBlockedClients(void) { - if(!unblocked_clients) return; - - while(listLength(unblocked_clients) > 0) { - // we need to check it every time, so that if one of the unblocked - // clients executed pause command, then we stop processing further. - if (isPausedActionsWithUpdate(PAUSE_ACTIONS_CLIENT_ALL_SET)) return; - listNode *ln = listFirst(unblocked_clients); - serverAssert(ln != NULL); - client *c = listNodeValue(ln); - listDelNode(unblocked_clients, ln); - c->flag.blockInuse_unblocked = 0; - total_clients_resumed_lifetime++; - // make the fd readble again. This should succeed as we are not adding a - // new client. If it fails because epoll_ctl failed then freeClient. - // We avoid setting the read handler for fake client that does not have a connection. - if (c->conn && connSetReadHandler(c->conn, readQueryFromClient) == C_ERR) { - freeClient(c); - return; - } - processBlockedCommand(c); +// Harry check TODO +int blockInuse_processUnblockClients(client *c) { + /* Process all the commands for an unblocked client. First we read the blocked command which is already parsed by calling processCommand. + * Then we process the commands present in querybuf by calling processInputBuffer. + */ + if (c->flag.close_asap) return; + c->flag.pending_command = 0; + int retval = processCommandAndResetClient(c); + if (retval != C_OK || blockInuse_isBlockedClient(c)) { + return; + } + //process the pending commands in the buffer. + if (processInputBuffer(c) == C_OK && !c->flag.close_asap) { + beforeNextClient(c); } } // Harry check Done -// remove a client from everywhere +// remove a client from the tables, the client must be blocked before calling void blockInuse_unlinkClient(client *c) { - serverAssert(blockInuse_isBlockedClient(c) || blockInuse_isUnblockedClient(c)); - if (blockInuse_isBlockedClient(c)) { - blockInuse_clientMetadata *metadata = getClientMetadata(c); - if (metadata == NULL) return; // return immediately if the client was not blocked on any keys. - unlinkBlockedClientOnKeys(c); - c->flag.blockInuse_blocked = 0; - removeClientMetadata(c); // remove the global hashtable entry - } + serverAssert(blockInuse_isBlockedClient(c)); + blockInuse_clientMetadata *metadata = getClientMetadata(c); + if (metadata == NULL) return; // return immediately if the client was not blocked on any keys. - if (blockInuse_isUnblockedClient(c)) { - listNode *ln = listSearchKey(unblocked_clients, c); - serverAssert(ln != NULL); - listDelNode(unblocked_clients, ln); - c->flag.blockInuse_unblocked = 0; - total_clients_resumed_lifetime++; - } + // remove from key_to_clients + unlinkBlockedClientOnKeys(c); + + // remove from client_to_keys + c->flag.blockInuse_blocked = 0; + removeClientMetadata(c); // remove the global hashtable entry } diff --git a/src/blocked_inuse.h b/src/blocked_inuse.h index fdb8f26aa96..e45b4296273 100644 --- a/src/blocked_inuse.h +++ b/src/blocked_inuse.h @@ -1,9 +1,31 @@ /* -Harry TODO: -1. write a brief introduction here -2. remove unblocked client once done -3. do we need the oldest? -*/ + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + * + * Client blocking mechanism for keys currently in use by other operations. + * + * This module provides a specialized blocking system that prevents concurrent access to keys + * that are actively being modified or processed. Unlike the generic blocking operations in + * blocked.c (BLPOP, WAIT, etc.), this mechanism blocks clients when they attempt to access + * keys that are marked as "in use" by internal operations such as bgIteration. + * + * Key features: + * - Blocks clients on multiple keys simultaneously + * - Automatically unblocks clients when all their requested keys become available + * - Maintains bidirectional mappings: client->keys and key->clients + * - Integrates with the server's event loop via processServerBlockedClients() + * - Tracks blocking statistics and lifetime metrics + * + * Typical workflow: + * 1. blockInuse_blockClientOnKeys() - Block a client on a set of keys + * 2. Keys remain blocked until explicitly unblocked + * 3. blockInuse_unblockClientsOnKey() - Unblock specific key, triggering client resumption + * 4. blockInuse_processServerBlockedClients() - Process unblocked clients in beforeSleep() + * + * This is used to ensure data consistency during operations that require exclusive access + * to keys, preventing race conditions and maintaining transactional integrity. + */ #ifndef BLOCKED_INUSE_H__ #define BLOCKED_INUSE_H__ @@ -12,28 +34,19 @@ Harry TODO: #include "adlist.h" struct robj; //defined in server.h -struct serverCommand; //defined in server.h struct client; //defined in server.h -/* Check if client is blocked/unblocked by blockInuse */ +/* Check if client is blocked by blockInuse */ #define blockInuse_isBlockedClient(c) ((c)->flag.blockInuse_blocked) -#define blockInuse_isUnblockedClient(c) ((c)->flag.blockInuse_unblocked) - -// Harry Check: do I need them in here, since they are totally private now. -typedef struct blockInuse_clientMetadata blockInuse_clientMetadata; -typedef struct blockInuse_blockingInfo blockInuse_blockingInfo; -/* Initialize global blockInuse structures. Call once at server startup. */ +/* Initialize blockInuse structures. Call once at server startup. */ void blockInuse_init(void); -/* If no clients are blocked and all the previously blocked clients have processed the blocked command, - then this will free the data structures and return true. Otherwise, this will return false, and we - still have few clients to process and cleanup failed. */ -bool blockInuse_cleanDbBlockingInfo(void); +/* Free blockInuse data structures, no clients should be blocked by blockInuse at this time. */ +void blockInuse_release(void); /* Returns the total count of currently blocked clients by blockInuse */ int blockInuse_getNumberOfBlockedClients(void); -long blockInuse_getNumberOfUnblockedClients(void); /* * Block given client on set of keys. Duplicated keys are handled. diff --git a/src/networking.c b/src/networking.c index 3c1dde147b8..9a66272ef9a 100644 --- a/src/networking.c +++ b/src/networking.c @@ -1986,8 +1986,8 @@ void unlinkClient(client *c) { /* Clear the tracking status. */ if (c->flag.tracking) disableTracking(c); - // We should never have a client here which is in blockInuse unblocked or blocked state. - serverAssert(!(blockInuse_isBlockedClient(c) || blockInuse_isUnblockedClient(c))); + // We should never have a client here which is in unblocked or blockInuse blocked state. + serverAssert(!(blockInuse_isBlockedClient(c) || (c)->flag.unblocked)); } /* Clear the client state to resemble a newly connected client. */ diff --git a/src/server.c b/src/server.c index 87807b7c968..ebb29bb65e5 100644 --- a/src/server.c +++ b/src/server.c @@ -1981,10 +1981,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) { /* Close clients that need to be closed asynchronous */ freeClientsInAsyncFreeQueue(); - if (blockInuse_getNumberOfUnblockedClients() > 0) { - blockInuse_processServerBlockedClients(); - } - /* Incrementally trim replication backlog, 10 times the normal speed is * to free replication backlog as much as possible. */ if (server.repl_backlog) incrementalTrimReplicationBacklog(10 * REPL_BACKLOG_TRIM_BLOCKS_PER_CALL); @@ -4226,7 +4222,7 @@ void unprepareCommand(client *c) { * if C_ERR is returned the client was destroyed (i.e. after QUIT). */ int processCommand(client *c) { - serverAssert(!(blockInuse_isBlockedClient(c) || blockInuse_isUnblockedClient(c))); + serverAssert(!(blockInuse_isBlockedClient(c) || c->flag.unblocked == 1)); if (!scriptIsTimedout()) { /* Both EXEC and scripts call call() directly so there should be @@ -4877,7 +4873,7 @@ int finishShutdown(void) { closeListeningSockets(1); /* Cleanup blockInuse data structures */ - blockInuse_cleanDbBlockingInfo(); + blockInuse_release(); moduleUnloadAllModules(); From 476c5c87fb2ffeb8fb8065077837e4672e98f3f4 Mon Sep 17 00:00:00 2001 From: Harry Lin Date: Wed, 4 Feb 2026 13:46:55 -0800 Subject: [PATCH 3/4] Format code --- cmake/Modules/SourceFiles.cmake | 2 +- src/blocked.c | 15 +-- src/blocked_inuse.c | 220 ++++++++++++++++++-------------- src/blocked_inuse.h | 103 +++++++++------ src/networking.c | 10 +- src/server.c | 89 ++++++++----- src/server.h | 1 - src/timeout.c | 9 +- 8 files changed, 258 insertions(+), 191 deletions(-) diff --git a/cmake/Modules/SourceFiles.cmake b/cmake/Modules/SourceFiles.cmake index 8cd97cd5b01..699146024c4 100644 --- a/cmake/Modules/SourceFiles.cmake +++ b/cmake/Modules/SourceFiles.cmake @@ -64,7 +64,7 @@ set(VALKEY_SERVER_SRCS ${CMAKE_SOURCE_DIR}/src/notify.c ${CMAKE_SOURCE_DIR}/src/setproctitle.c ${CMAKE_SOURCE_DIR}/src/blocked.c - ${CMAKE_SOURCE_DIR}/src/server_blocking.c + ${CMAKE_SOURCE_DIR}/src/blocked_inuse.c ${CMAKE_SOURCE_DIR}/src/hyperloglog.c ${CMAKE_SOURCE_DIR}/src/latency.c ${CMAKE_SOURCE_DIR}/src/sparkline.c diff --git a/src/blocked.c b/src/blocked.c index 3d9b61f6522..d69bc9c0c9d 100644 --- a/src/blocked.c +++ b/src/blocked.c @@ -160,8 +160,7 @@ void processUnblockedClients(void) { client *c; while (listLength(server.unblocked_clients)) { - // we need to check it every time, so that if one of the unblocked - // clients executed pause command, then we stop processing further. + // If one of the unblocked clients executed pause command, then we stop processing further. if (isPausedActionsWithUpdate(PAUSE_ACTIONS_CLIENT_ALL_SET)) return; ln = listFirst(server.unblocked_clients); serverAssert(ln != NULL); @@ -176,28 +175,24 @@ void processUnblockedClients(void) { continue; } - if (blockInuse_isBlockedClient(c)) { - c->flag.blockInuse_unblocked = 0; - // make the fd readble again. This should succeed as we are not adding a - // new client. If it fails because epoll_ctl failed then freeClient. - // We avoid setting the read handler for fake client that does not have a connection. + if (blockInuse_clientBlocked(c)) { + // Enable the read handler. If it fails because epoll_ctl failed then freeClient. if (c->conn && connSetReadHandler(c->conn, readQueryFromClient) == C_ERR) { freeClient(c); return; } - if (c->flag.close_asap) return; // maybe move out of the if? } /* Process remaining data in the input buffer, unless the client * is blocked again. Actually processInputBuffer() checks that the * client is not blocked before to proceed, but things may change and * the code is conceptually more correct this way. */ - if (!c->flag.blocked && !blockInuse_isBlockedClient(c)) { + if (!c->flag.blocked && !blockInuse_clientBlocked(c)) { /* If we have a queued command, execute it now. */ if (processPendingCommandAndInputBuffer(c) == C_ERR) { continue; } } - if (c) beforeNextClient(c); + if (c && !c->flag.close_asap) beforeNextClient(c); } } diff --git a/src/blocked_inuse.c b/src/blocked_inuse.c index 40aa1c68d9d..ead27708982 100644 --- a/src/blocked_inuse.c +++ b/src/blocked_inuse.c @@ -1,3 +1,12 @@ +/* + * Copyright (c) Valkey Contributors + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + * + * BlockInuse - Client blocking mechanism for keys that are currently + * in use by other operations. + */ + #include "server.h" #include "blocked_inuse.h" @@ -8,19 +17,19 @@ extern uint64_t hashtableClientHash(const void *key); extern int hashtableClientKeyCompare(const void *key1, const void *key2); // Internal blockInuse data structure -static hashtable *client_to_keys; /* Map(client memory address) -> blockInuse_clientMetadata (contains a list of keys). */ -static hashtable *key_to_clients; /* Map(key)->list(blocked clients). */ -uint32_t blocked_clients_on_keys; /* Num of clients blocked on keys now. */ -uint32_t blockInuse_total_clients_blocked_on_keys_lifetime; /* Num of clients blocked on keys lifetime. */ -uint32_t blockInuse_total_clients_unblocked_on_keys_lifetime; /* Num of clients unblocked on keys lifetime. */ +static hashtable *client_to_keys; /* Maps client pointers to blockInuse_clientMetadata (list of keys). */ +static hashtable *key_to_clients; /* Maps keys to a list of clients blocked on them. */ +static uint32_t blocked_clients_on_keys; /* Current number of clients blocked on keys. */ +/* Metadata for a blocked client */ typedef struct blockInuse_clientMetadata { - robj **keys; /* Keys this client is blocked on. */ - int n_keys; /* Number of keys this client is blocked on. */ - long long blocked_at; /* What time this client is blocked in milli second. It uses value from server.mstime. */ + robj **keys; /* Array of keys the client is blocked on */ + int n_keys; /* Number of keys in the array */ + long long blocked_at; /* Timestamp when client was blocked (from server.mstime) */ } blockInuse_clientMetadata; /* ----------------------------- client_to_keys Hashtable util ------------------------- */ +/* Entry for client_to_keys hashtable */ typedef struct { client *c; blockInuse_clientMetadata metadata; @@ -34,8 +43,11 @@ static const void *clientDataEntryGetKey(const void *entry) { static void clientDataEntryDestructor(void *entry) { clientDataEntry *e = entry; if (e->metadata.keys) { - //Refcount for all the keys should be decreased before calling this function. Hence n_keys should be 0. - serverAssert(e->metadata.keys == 0); + // Refcounts for all keys should be decreased before calling this function + // Ensure that n_keys is 0 before freeing keys + serverAssert(e->metadata.n_keys == 0); + + // Decrease refcount for each key (defensive, in case n_keys > 0) for (int i = 0; i < e->metadata.n_keys; i++) { decrRefCount(e->metadata.keys[i]); } @@ -52,6 +64,8 @@ static hashtableType clientDataHashtableType = { }; /* Utility functions for client_to_keys hashtable */ + +// Find metadata for a client; create if not found static blockInuse_clientMetadata *addOrFindClientMetadata(client *c) { clientDataEntry *entry; if (hashtableFind(client_to_keys, c, (void **)&entry)) { @@ -64,6 +78,7 @@ static blockInuse_clientMetadata *addOrFindClientMetadata(client *c) { return &entry->metadata; } +// Get metadata for a client; return NULL if not found static blockInuse_clientMetadata *getClientMetadata(client *c) { clientDataEntry *entry; if (hashtableFind(client_to_keys, c, (void **)&entry)) { @@ -72,18 +87,21 @@ static blockInuse_clientMetadata *getClientMetadata(client *c) { return NULL; } +// Remove a client and its metadata from the hashtable static void removeClientMetadata(client *c) { hashtablePop(client_to_keys, c, NULL); } - /* ----------------------------- key_to_clients Hashtable Util ------------------------- */ -/* Entry type for key_to_clients: robj key -> list of clients */ + +/* Hashtable entry: maps a key object to the list of clients blocked on it */ typedef struct { - robj *key; - list *clients; + robj *key; /* Key object */ + list *clients; /* List of clients blocked on this key */ } keyToClientsEntry; +/* Hashtable callbacks */ + static const void *keyToClientsGetKey(const void *entry) { return ((keyToClientsEntry *)entry)->key; } @@ -103,6 +121,8 @@ static hashtableType keyToClientsHashtableType = { }; /* Utility functions for key_to_clients hashtable */ + +// Get or create the list of clients blocked on a key static list *addOrFindBlockedClientsListsByKey(robj *key) { keyToClientsEntry *entry; if (hashtableFind(key_to_clients, key, (void **)&entry)) { @@ -117,6 +137,7 @@ static list *addOrFindBlockedClientsListsByKey(robj *key) { return entry->clients; } +// Get the list of clients blocked on a key, or NULL if none static list *getBlockedClientsListsByKey(robj *key) { keyToClientsEntry *entry; if (hashtableFind(key_to_clients, key, (void **)&entry)) { @@ -125,21 +146,29 @@ static list *getBlockedClientsListsByKey(robj *key) { return NULL; } +// Remove a key and its client list from the hashtable static void removeBlockedClientsListsByKey(robj *key) { hashtablePop(key_to_clients, key, NULL); } /* ----------------------------- util ------------------------- */ + static void markClientBlocked(client *c) { c->flag.blockInuse_blocked = 1; - c->flag.pending_command = 1; // Harrt TODO do we need this, and why? + c->flag.pending_command = 1; } -// Init the client Metadata and insert it into client_to_keys hash table. +/* + * Initialize metadata for a client and insert it into client_to_keys hashtable. + * + * nKeys specifies the size for the keys array. + * Returns the pointer to the initialized metadata. + */ static blockInuse_clientMetadata *initClientMetadata(client *c, int nKeys) { - serverAssert(!getClientMetadata(c)); // this client must not be in our global table - serverAssert(nKeys >= 0); // non negative check - blockInuse_clientMetadata *metadata = addOrFindClientMetadata(c); // add metada into the hashtable + serverAssert(!getClientMetadata(c)); // client must not already exist + serverAssert(nKeys >= 0); + + blockInuse_clientMetadata *metadata = addOrFindClientMetadata(c); metadata->n_keys = 0; metadata->keys = NULL; metadata->blocked_at = server.mstime; @@ -149,10 +178,14 @@ static blockInuse_clientMetadata *initClientMetadata(client *c, int nKeys) { return metadata; } -// Remove this client totally from every lists in the key_to_clients table. +/* + * Unlink a blocked client from all key_to_clients entries + * and release references in its client metadata. + */ static void unlinkBlockedClientOnKeys(client *c) { blockInuse_clientMetadata *metadata = getClientMetadata(c); if (!metadata) return; + for (int i = 0; i < metadata->n_keys; ++i) { robj *key = metadata->keys[i]; @@ -161,18 +194,22 @@ static void unlinkBlockedClientOnKeys(client *c) { listDelNode(clientList, listSearchKey(clientList, c)); if (listLength(clientList) == 0) removeBlockedClientsListsByKey(key); - decrRefCount(key); // Harry TODO: double check where to increase and where to decrease. + + decrRefCount(key); metadata->keys[i] = NULL; } metadata->n_keys = 0; blocked_clients_on_keys--; - blockInuse_total_clients_unblocked_on_keys_lifetime++; } -// Remove a key from a client entry in client_to_keys table + +/* + * Remove a specific key from a client's blocked keys array. + */ static blockInuse_clientMetadata *removeBlockingKeyFromClient(client *c, robj *key) { blockInuse_clientMetadata *metadata = getClientMetadata(c); if (metadata == NULL) return NULL; + sds key_sds = objectGetKey(key); for (int i = 0; i < metadata->n_keys; ++i) { sds curr_key = objectGetKey(metadata->keys[i]); @@ -184,115 +221,127 @@ static blockInuse_clientMetadata *removeBlockingKeyFromClient(client *c, robj *k return metadata; } } - // we expect to find a key - serverAssert(false); + serverAssert(false); // key must exist } /* ----------------------------- API implementation ------------------------- */ -/* Initialize global client_to_keys hashtable. Call once at server startup. */ -/* Initializes the blockInuse data structures needed for DB. Called at server startup per db. */ -// Harry check Done +/* + * Initialize blockInuse data structures. + */ void blockInuse_init(void) { client_to_keys = hashtableCreate(&clientDataHashtableType); key_to_clients = hashtableCreate(&keyToClientsHashtableType); blocked_clients_on_keys = 0; - blockInuse_total_clients_blocked_on_keys_lifetime = 0; - blockInuse_total_clients_unblocked_on_keys_lifetime = 0; } -/* Clean up the blockInuse data structures for the database if possible. - * Returns false if there are any blocked or unblocked clients (no cleanup performed). - * Returns true if cleanup succeeded or was already done. - * Note: This will not free the struct itself but cleans up the internal data. */ +/* + * Release blockInuse data structures. + * Only allowed if no clients are currently blocked. + */ void blockInuse_release(void) { serverAssert(blocked_clients_on_keys == 0); - hashtableRelease(key_to_clients); - key_to_clients = NULL; + + if (client_to_keys) { + hashtableRelease(client_to_keys); + client_to_keys = NULL; + } + if (key_to_clients) { + hashtableRelease(key_to_clients); + key_to_clients = NULL; + } blocked_clients_on_keys = 0; - return true; } -// Harry check Done +/* Get the current number of clients blocked by blockInuse. */ int blockInuse_getNumberOfBlockedClients(void) { return blocked_clients_on_keys; } -// Harry check Done +/* Block a client on a set of keys. */ int blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]) { - // some checks - serverAssert(!(blockInuse_isBlockedClient(c) || (c)->flag.unblocked)); + // Ensure client is not already blocked or unblocked + serverAssert(!(blockInuse_clientBlocked(c) || (c)->flag.unblocked)); + if (nKeys == 0) return C_ERR; - if (c->flag.replica) return C_ERR; // Maybe remove this? + if (c->flag.replica) return C_ERR; for (int i = 0; i < nKeys; ++i) { if (keys[i]->type != OBJ_STRING) return C_ERR; } - // add into the global table - blockInuse_clientMetadata *metadata = initClientMetadata(c, nKeys); // this will add into the global table and assign memory, but the keys in metadata is still empty + // Initialize client metadata and insert into client_to_keys table + blockInuse_clientMetadata *metadata = initClientMetadata(c, nKeys); markClientBlocked(c); + for (int i = 0; i < nKeys; ++i) { - // This loop is just for book keeping, we want to do 2 things in the loop: - // 1. adding the key into the client metadata (global table), such that c -> list of keys (append key to tail) - // 2. add the entry into the blocking info, such that key -> list of clients (append c to tail) + robj *key = keys[i]; - list *blockedClientsList = addOrFindBlockedClientsListsByKey(keys[i]); // this will add entry of key -> [client 1, client 2 ...], incrRefCount in included. + // Get or create the list of clients blocked on this key + list *blockedClientsList = addOrFindBlockedClientsListsByKey(key); - // If the last client blocked on this key is not c, then we add c onto the list. - // Otherwise this is a duplicated key and we should ignore it. + // Deduplicate: add client only if it’s not already the last in the list listNode *last_client = listLast(blockedClientsList); - - // this if check is for deduplica of keys if (last_client == NULL || last_client->value != c) { - // 2. add the client into the blocking info at tail, either create a new entry or add to tail + // Add client to the key’s blocked clients list listAddNodeTail(blockedClientsList, c); - // 1. add the key into the client metadata, so it would be client -> [key1, key2, key3] - incrRefCount(keys[i]); - metadata->keys[metadata->n_keys] = keys[i]; + // Add key to the client’s metadata and increment reference count + incrRefCount(key); + metadata->keys[metadata->n_keys] = key; metadata->n_keys++; } } + // Disable client’s Read Handler to prevent reading commands while blocked if (c->conn) { - // Delete the readable event from the event loop for the blocked client. connSetReadHandler(c->conn, NULL); } blocked_clients_on_keys++; - blockInuse_total_clients_blocked_on_keys_lifetime++; return C_OK; } -/* Unblock given key. A client will be unblocked, if it has no more dependency on any key and will be put into unblocked_clients list. */ -// Harry check Done +/* + * Unblock all clients blocked on the given key. + * + * - Each client is unblocked only when it has no remaining dependencies on other keys. + * - Clients that become fully unblocked are added to server.unblocked_clients + * and will be resumed later in processUnblockedClients(). + */ void blockInuse_unblockClientsOnKey(robj *key) { list *blockedClientsList = getBlockedClientsListsByKey(key); if (blockedClientsList == NULL) return; + serverAssert(listLength(blockedClientsList) > 0); + while (listLength(blockedClientsList) > 0) { listNode *ln = listFirst(blockedClientsList); client *c = listNodeValue(ln); + + // Remove client from this key's blocked list listDelNode(blockedClientsList, ln); - // remove a key for a specific client in client_to_keys + + // Remove this key from the client's blocked key list blockInuse_clientMetadata *metadata = removeBlockingKeyFromClient(c, key); + if (metadata->n_keys == 0) { - // time to remove this entry in our global table + // Client has no more blocked keys → mark unblocked serverAssert(c->flag.unblocked == 0); - if (!c->flag.unblocked) { - c->flag.unblocked = 1; - listAddNodeTail(server.unblocked_clients, c); - } + c->flag.unblocked = 1; + listAddNodeTail(server.unblocked_clients, c); + + // Remove client metadata from client_to_keys table removeClientMetadata(c); blocked_clients_on_keys--; - blockInuse_total_clients_blocked_on_keys_lifetime++; } } - // remove from key_to_clients + // Remove the key entry from key_to_clients table removeBlockedClientsListsByKey(key); } -// Harry check Done +/* + * Unblock all clients on all keys. + */ void blockInuse_unblockClientsOnAllKeys(void) { hashtableIterator iter; hashtableInitIterator(&iter, key_to_clients, HASHTABLE_ITER_SAFE); @@ -307,34 +356,19 @@ void blockInuse_unblockClientsOnAllKeys(void) { hashtableCleanupIterator(&iter); } -// Harry check TODO -int blockInuse_processUnblockClients(client *c) { - /* Process all the commands for an unblocked client. First we read the blocked command which is already parsed by calling processCommand. - * Then we process the commands present in querybuf by calling processInputBuffer. - */ - if (c->flag.close_asap) return; - c->flag.pending_command = 0; - int retval = processCommandAndResetClient(c); - if (retval != C_OK || blockInuse_isBlockedClient(c)) { - return; - } - //process the pending commands in the buffer. - if (processInputBuffer(c) == C_OK && !c->flag.close_asap) { - beforeNextClient(c); - } -} - -// Harry check Done -// remove a client from the tables, the client must be blocked before calling +/* + * Unlink a blocked client from all blockInuse structures, the client must be blocked by blockInuse. + */ void blockInuse_unlinkClient(client *c) { - serverAssert(blockInuse_isBlockedClient(c)); + serverAssert(blockInuse_clientBlocked(c) && c->flag.unblocked == 0); + blockInuse_clientMetadata *metadata = getClientMetadata(c); - if (metadata == NULL) return; // return immediately if the client was not blocked on any keys. + if (metadata == NULL) return; // Client has no blocking metadata - // remove from key_to_clients + // Remove client from all key-to-client lists unlinkBlockedClientOnKeys(c); - // remove from client_to_keys + // Clear the blocked flag and remove client metadata c->flag.blockInuse_blocked = 0; - removeClientMetadata(c); // remove the global hashtable entry + removeClientMetadata(c); } diff --git a/src/blocked_inuse.h b/src/blocked_inuse.h index e45b4296273..cc551a19c1d 100644 --- a/src/blocked_inuse.h +++ b/src/blocked_inuse.h @@ -3,28 +3,34 @@ * All rights reserved. * SPDX-License-Identifier: BSD-3-Clause * + * * Client blocking mechanism for keys currently in use by other operations. * - * This module provides a specialized blocking system that prevents concurrent access to keys - * that are actively being modified or processed. Unlike the generic blocking operations in - * blocked.c (BLPOP, WAIT, etc.), this mechanism blocks clients when they attempt to access - * keys that are marked as "in use" by internal operations such as bgIteration. + * This module provides a specialized blocking system that prevents concurrent + * access to keys that are actively being modified or processed. Unlike the + * generic blocking operations in blocked.c, this mechanism blocks clients when + * they attempt to access keys that are marked as "in use" by internal + * operations such as bgIteration. * * Key features: - * - Blocks clients on multiple keys simultaneously - * - Automatically unblocks clients when all their requested keys become available - * - Maintains bidirectional mappings: client->keys and key->clients - * - Integrates with the server's event loop via processServerBlockedClients() - * - Tracks blocking statistics and lifetime metrics + * - Blocks clients on multiple keys simultaneously + * - Automatically unblocks clients when all their requested keys become + * available + * - Maintains bidirectional mappings: client->keys and key->clients + * - Integrates with the server's event loop via processUnblockedClients() + * in blocked.c + * + * Workflow: + * 1. blockInuse_blockClientOnKeys() - Block a client on a set of keys + * 2. Keys remain blocked until explicitly unblocked + * 3. blockInuse_unblockClientsOnKey() - Unblock specific key, triggering + * clients resumption + * 4. processUnblockedClients() - Process unblocked clients in beforeSleep() * - * Typical workflow: - * 1. blockInuse_blockClientOnKeys() - Block a client on a set of keys - * 2. Keys remain blocked until explicitly unblocked - * 3. blockInuse_unblockClientsOnKey() - Unblock specific key, triggering client resumption - * 4. blockInuse_processServerBlockedClients() - Process unblocked clients in beforeSleep() + * This is used to ensure data consistency during operations that require + * exclusive access to keys, preventing race conditions and maintaining + * transactional integrity. * - * This is used to ensure data consistency during operations that require exclusive access - * to keys, preventing race conditions and maintaining transactional integrity. */ #ifndef BLOCKED_INUSE_H__ @@ -33,50 +39,63 @@ #include "hashtable.h" #include "adlist.h" -struct robj; //defined in server.h -struct client; //defined in server.h +struct robj; // defined in server.h +struct client; // defined in server.h /* Check if client is blocked by blockInuse */ -#define blockInuse_isBlockedClient(c) ((c)->flag.blockInuse_blocked) +#define blockInuse_clientBlocked(c) ((c)->flag.blockInuse_blocked) -/* Initialize blockInuse structures. Call once at server startup. */ +/* Initialize blockInuse structures, must be called once during server startup. */ void blockInuse_init(void); -/* Free blockInuse data structures, no clients should be blocked by blockInuse at this time. */ +/* Free blockInuse data structures, no clients must be blocked by blockInuse. */ void blockInuse_release(void); -/* Returns the total count of currently blocked clients by blockInuse */ +/* Return the number of clients currently blocked by blockInuse. */ int blockInuse_getNumberOfBlockedClients(void); /* - * Block given client on set of keys. Duplicated keys are handled. - * To avoid the extra copy, we keep reference to the passed keys. So passed variable keys, should be heap allocated. - * API asserts that the client do not already have a blocked/unblocked flag set. - * Return Value: - * C_ERR: if - * a. Any passed key is not sds. - * b. nKeys = 0. - * c. Client is slave client. - * Otherwise, it blocks the client and returns C_OK. - * */ -// Blocks a client on a set of keys. -// Then client will remain blocked until all keys are unblocked. + * Block a client on a set of keys. Duplicate keys are allowed and handled. + * + * To avoid extra copying, this API keeps references to the passed key objects. + * The caller must ensure that the `keys` array and all key objects are + * heap-allocated and remain valid for the duration of the block. + * + * Preconditions: + * - The client must not already have any blocked or unblocked flags set. + * + * Return value: + * - C_ERR if: + * a. nKeys == 0 + * b. any key is not a sds string object + * c. the client is a replica client + * - C_OK otherwise; the client is blocked until all keys are unblocked. + */ int blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]); -/* Unblock given key. A client will be unblocked if it has no more dependency on any key and will be - * put into unblocked_clients list. Clients from this list are processed during processUnblockedClients. +/* + * Unblock clients blocked on the given key. + * + * A client is unblocked only when it has no remaining dependencies on any + * blocked keys. Such clients are added to the server.unblocked_clients list and + * resumed later during processUnblockedClients() in blocked.c. */ void blockInuse_unblockClientsOnKey(robj *key); -/* Unblock all clients on all keys */ +/* + * Unblock all clients blocked by blockInuse on all keys. + * + * Clients that become unblocked are added to the server.unblocked_clients + * list and resumed later during processUnblockedClients(). + */ void blockInuse_unblockClientsOnAllKeys(void); -/* If clientBlocking is enabled, this function is called in beforeSleep each time, to resume clients which were previously blocked. */ -void blockInuse_processServerBlockedClients(void); - /* - * This API is to force unlinking of a blocked client. Typically required when we want to free the client while its blocked (e.g. memory pressure). - * This will clean up the current command arguments and detach all the references in blocking structures. + * Unlink a client currently blocked by blockInuse. Typically used when + * a client is being freed while still blocked (e.g., due to memory pressure). + * + * This function removes the client from all blockInuse data structures + * and clears its blockInuse blocked flag. */ void blockInuse_unlinkClient(client *c); diff --git a/src/networking.c b/src/networking.c index 7dbb6c2dedd..229ed03537f 100644 --- a/src/networking.c +++ b/src/networking.c @@ -1904,7 +1904,7 @@ void unlinkClient(client *c) { /* If this is marked as current client unset it. */ if (c->conn && server.current_client == c) server.current_client = NULL; - blockInuse_unlinkClient(c); + if (blockInuse_clientBlocked(c)) blockInuse_unlinkClient(c); /* Certain operations must be done only if the client has an active connection. * If the client was already unlinked or if it's a "fake client" the @@ -1990,8 +1990,8 @@ void unlinkClient(client *c) { /* Clear the tracking status. */ if (c->flag.tracking) disableTracking(c); - // We should never have a client here which is in unblocked or blockInuse blocked state. - serverAssert(!(blockInuse_isBlockedClient(c) || (c)->flag.unblocked)); + // Here client should never in unblocked or blockInuse blocked state. + serverAssert(!(blockInuse_clientBlocked(c) || (c)->flag.unblocked)); } /* Clear the client state to resemble a newly connected client. */ @@ -3808,6 +3808,7 @@ int processPendingCommandAndInputBuffer(client *c) { * But in case of a module blocked client (see RM_Call 'K' flag) we do not reach this code path. * So whenever we change the code here we need to consider if we need this change on module * blocked client as well */ + if (c->flag.close_asap) return C_ERR; if (c->flag.pending_command) { c->flag.pending_command = 0; if (processCommandAndResetClient(c) == C_ERR) { @@ -4266,7 +4267,7 @@ int isClientConnIpV6(client *c) { * readable format, into the sds string 's'. */ sds catClientInfoString(sds s, client *client, int hide_user_data) { if (!server.crashed) waitForClientIO(client); - char flags[17], events[3], capa[9], conninfo[CONN_INFO_LEN], *p; + char flags[18], events[3], capa[9], conninfo[CONN_INFO_LEN], *p; p = flags; if (client->flag.replica) { @@ -4280,6 +4281,7 @@ sds catClientInfoString(sds s, client *client, int hide_user_data) { if (client->flag.pubsub) *p++ = 'P'; if (client->flag.multi) *p++ = 'x'; if (client->flag.blocked) *p++ = 'b'; + if (client->flag.blockInuse_blocked) *p++ = 'X'; if (client->flag.tracking) *p++ = 't'; if (client->flag.tracking_broken_redir) *p++ = 'R'; if (client->flag.tracking_bcast) *p++ = 'B'; diff --git a/src/server.c b/src/server.c index da55be6a5cc..e35365a10c4 100644 --- a/src/server.c +++ b/src/server.c @@ -76,6 +76,9 @@ #include #include #include +#ifdef __APPLE__ +#include +#endif #ifdef __linux__ #include @@ -1155,40 +1158,57 @@ void getExpensiveClientsInfo(size_t *in_usage, size_t *out_usage) { *out_usage = o; } -// return 1 if client was terminated, 0 if still alive. +/* + * Check if a blockInuse blocked client connection has been closed from the remote side. + * + * Returns: + * 1 if the client has been terminated, + * 0 if still alive. + */ static int clientsCronCheckBlockInuseClients(client *c) { - // Check for clients with no read/write handlers (blocked clients) - // which have been closed from the remote side. - if (c->conn) { - // It's a normal client connection (not a fake client) ... - if (c->conn->type == connectionTypeTcp() || c->conn->type == connectionTypeTls()) { - // ... and it's based on a TCP socket ... - if (aeGetFileEvents(server.el, c->conn->fd) == AE_NONE) { - // ... and neither read nor write handler is installed ... - // Determine if the connection has been closed, from the far end, by - // checking the TCP state information. - struct tcp_info info; - socklen_t infolen = sizeof(info); - // Query the kernal for TCP socket state info - // since no event handler exists, we must manally check if the connection is dead. - if (getsockopt(c->conn->fd, IPPROTO_TCP, TCP_INFO, &info, &infolen) == 0) { - // check TCP state - if (info.tcpi_state == TCP_CLOSE_WAIT || info.tcpi_state == TCP_CLOSE) { - // TCP_CLOSE_WAIT: remote side closed, local side hasn't closed yet - // TCP_CLOSE: connection fully closed. - if (server.verbosity <= LL_VERBOSE) { - sds info = catClientInfoString(sdsempty(), c, server.hide_user_data_from_log); - serverLog(LL_VERBOSE, "Client closed connection while blocked %s", info); - sdsfree(info); - } - freeClientAsync(c); - return 1; // client has been closed - } - } - } + if (!c->conn) return 0; // No connection, cannot check + + // Only TCP or TLS clients are relevant + if (c->conn->type != connectionTypeTcp() && c->conn->type != connectionTypeTls()) return 0; + + // If neither read nor write handler is installed, client is blocked. + if (aeGetFileEvents(server.el, c->conn->fd) != AE_NONE) return 0; + + // No event handler exists, check TCP socket state. + + /* TCP state introspection is platform-specific: + * - Linux: TCP_INFO / struct tcp_info + * - macOS: TCP_CONNECTION_INFO / struct tcp_connection_info (BSD TCP FSM) + */ +#if defined(__linux__) + struct tcp_info info; + socklen_t infolen = sizeof(info); + if (getsockopt(c->conn->fd, IPPROTO_TCP, TCP_INFO, &info, &infolen) != 0 || infolen < sizeof(info)) return 0; // Cannot retrieve TCP info + + if (info.tcpi_state == TCP_CLOSE_WAIT || + info.tcpi_state == TCP_CLOSE) +#elif defined(__APPLE__) + struct tcp_connection_info info; + socklen_t infolen = sizeof(info); + + if (getsockopt(c->conn->fd, IPPROTO_TCP, TCP_CONNECTION_INFO, &info, &infolen) != 0 || infolen < sizeof(info)) return 0; // Cannot retrieve TCP info + + // Check if connection is closed or half-closed + if (info.tcpi_state == TCPS_CLOSE_WAIT || + info.tcpi_state == TCPS_CLOSED) +#endif + { + if (server.verbosity <= LL_VERBOSE) { + sds client_info = catClientInfoString(sdsempty(), c, server.hide_user_data_from_log); + serverLog(LL_VERBOSE, "Client closed connection while blocked %s", client_info); + sdsfree(client_info); } + + freeClientAsync(c); + return 1; // Client has been closed } - return 0; // client has not been terminated + + return 0; // Client is still alive } /* This function is called by clientsTimeProc() and is used in order to perform @@ -2972,7 +2992,7 @@ void initServer(void) { server.debug_client_enforce_reply_list = 0; resetReplicationBuffer(); - /* Init blockInuse */ + /* Init blockInuse data structures */ blockInuse_init(); /* Make sure the locale is set on startup based on the config file. */ @@ -4257,8 +4277,7 @@ void unprepareCommand(client *c) { * other operations can be performed by the caller. Otherwise * if C_ERR is returned the client was destroyed (i.e. after QUIT). */ int processCommand(client *c) { - - serverAssert(!(blockInuse_isBlockedClient(c) || c->flag.unblocked == 1)); + serverAssert(!(blockInuse_clientBlocked(c) || c->flag.unblocked == 1)); if (!scriptIsTimedout()) { /* Both EXEC and scripts call call() directly so there should be @@ -4908,7 +4927,7 @@ int finishShutdown(void) { /* Close the listening sockets. Apparently this allows faster restarts. */ closeListeningSockets(1); - /* Cleanup blockInuse data structures */ + /* Release blockInuse data structures. */ blockInuse_release(); moduleUnloadAllModules(); diff --git a/src/server.h b/src/server.h index f92be966f22..28b2ef18382 100644 --- a/src/server.h +++ b/src/server.h @@ -1153,7 +1153,6 @@ typedef struct ClientFlags { uint64_t close_after_reply : 1; /* Close after writing entire reply. */ uint64_t unblocked : 1; /* This client was unblocked and is stored in server.unblocked_clients */ uint64_t blockInuse_blocked : 1; /* This client is blocked by blockInuse */ - uint64_t blockInuse_unblocked : 1; /* This client is unblocked by blockInuse */ uint64_t script : 1; /* This is a non connected client used by Lua */ uint64_t asking : 1; /* Client issued the ASKING command */ uint64_t close_asap : 1; /* Close this client ASAP */ diff --git a/src/timeout.c b/src/timeout.c index fb6c1ba8fe3..6d834e8cf36 100644 --- a/src/timeout.c +++ b/src/timeout.c @@ -55,12 +55,11 @@ int clientsCronHandleTimeout(client *c, mstime_t now_ms) { if (server.maxidletime && /* This handles the idle clients connection timeout if set. */ - !c->flag.replica && /* No timeout for replicas and monitors */ - !mustObeyClient(c) && /* No timeout for primaries and AOF */ - !c->flag.blocked && /* No timeout for BLPOP */ - !c->flag.pubsub && /* No timeout for Pub/Sub clients */ + !c->flag.replica && /* No timeout for replicas and monitors */ + !mustObeyClient(c) && /* No timeout for primaries and AOF */ + !c->flag.blocked && /* No timeout for BLPOP */ + !c->flag.pubsub && /* No timeout for Pub/Sub clients */ !c->flag.blockInuse_blocked && /* No timeout for BlockInuse client */ - !c->flag.blockInuse_unblocked && /* Client is unblocked, but we haven't yet processed the blocking command and input buffer, no timeout */ (now - c->last_interaction > server.maxidletime)) { serverLog(LL_VERBOSE, "Closing idle client"); freeClient(c); From 35a5a8224a2ec5d3387fe9ea1f98894a6c0fca67 Mon Sep 17 00:00:00 2001 From: Harry Lin Date: Fri, 13 Feb 2026 10:55:41 -0800 Subject: [PATCH 4/4] Address comments --- src/blocked.c | 13 ++++++----- src/blocked_inuse.c | 18 +++++++++------ src/blocked_inuse.h | 24 ++++---------------- src/networking.c | 7 +++--- src/server.c | 55 ++++++++++++++++++--------------------------- src/server.h | 1 - src/timeout.c | 11 ++++----- 7 files changed, 55 insertions(+), 74 deletions(-) diff --git a/src/blocked.c b/src/blocked.c index d69bc9c0c9d..8ef69751720 100644 --- a/src/blocked.c +++ b/src/blocked.c @@ -70,6 +70,7 @@ #include "monotonic.h" #include "cluster_slot_stats.h" #include "module.h" +#include "blocked_inuse.h" /* forward declarations */ static void unblockClientWaitingData(client *c); @@ -166,6 +167,7 @@ void processUnblockedClients(void) { serverAssert(ln != NULL); c = ln->value; listDelNode(server.unblocked_clients, ln); + serverAssert(!blockInuse_clientBlocked(c)); c->flag.unblocked = 0; if (c->flag.module) { @@ -175,9 +177,10 @@ void processUnblockedClients(void) { continue; } - if (blockInuse_clientBlocked(c)) { - // Enable the read handler. If it fails because epoll_ctl failed then freeClient. - if (c->conn && connSetReadHandler(c->conn, readQueryFromClient) == C_ERR) { + /* Reinstall read handler if it was removed (e.g. by blockInuse) */ + if (c->conn && !connHasReadHandler(c->conn)) { + // If it fails because epoll_ctl failed then freeClient. + if (connSetReadHandler(c->conn, readQueryFromClient) == C_ERR) { freeClient(c); return; } @@ -186,13 +189,13 @@ void processUnblockedClients(void) { * is blocked again. Actually processInputBuffer() checks that the * client is not blocked before to proceed, but things may change and * the code is conceptually more correct this way. */ - if (!c->flag.blocked && !blockInuse_clientBlocked(c)) { + if (!c->flag.blocked) { /* If we have a queued command, execute it now. */ if (processPendingCommandAndInputBuffer(c) == C_ERR) { continue; } } - if (c && !c->flag.close_asap) beforeNextClient(c); + if (!c->flag.close_asap) beforeNextClient(c); } } diff --git a/src/blocked_inuse.c b/src/blocked_inuse.c index ead27708982..91539a3da71 100644 --- a/src/blocked_inuse.c +++ b/src/blocked_inuse.c @@ -154,6 +154,7 @@ static void removeBlockedClientsListsByKey(robj *key) { /* ----------------------------- util ------------------------- */ static void markClientBlocked(client *c) { + serverAssert(c->flag.blocked == 0); c->flag.blockInuse_blocked = 1; c->flag.pending_command = 1; } @@ -226,6 +227,11 @@ static blockInuse_clientMetadata *removeBlockingKeyFromClient(client *c, robj *k /* ----------------------------- API implementation ------------------------- */ +/* Check if client is blocked by blockInuse */ +int blockInuse_clientBlocked(client *c) { + return c->flag.blockInuse_blocked; +} + /* * Initialize blockInuse data structures. */ @@ -259,14 +265,12 @@ int blockInuse_getNumberOfBlockedClients(void) { } /* Block a client on a set of keys. */ -int blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]) { - // Ensure client is not already blocked or unblocked +void blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]) { serverAssert(!(blockInuse_clientBlocked(c) || (c)->flag.unblocked)); - - if (nKeys == 0) return C_ERR; - if (c->flag.replica) return C_ERR; + serverAssert(nKeys > 0); + serverAssert(!c->flag.replica); for (int i = 0; i < nKeys; ++i) { - if (keys[i]->type != OBJ_STRING) return C_ERR; + serverAssert(keys[i]->type == OBJ_STRING); } // Initialize client metadata and insert into client_to_keys table @@ -297,7 +301,6 @@ int blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]) { connSetReadHandler(c->conn, NULL); } blocked_clients_on_keys++; - return C_OK; } /* @@ -327,6 +330,7 @@ void blockInuse_unblockClientsOnKey(robj *key) { // Client has no more blocked keys → mark unblocked serverAssert(c->flag.unblocked == 0); c->flag.unblocked = 1; + c->flag.blockInuse_blocked = 0; listAddNodeTail(server.unblocked_clients, c); // Remove client metadata from client_to_keys table diff --git a/src/blocked_inuse.h b/src/blocked_inuse.h index cc551a19c1d..57c4ddb1505 100644 --- a/src/blocked_inuse.h +++ b/src/blocked_inuse.h @@ -36,14 +36,11 @@ #ifndef BLOCKED_INUSE_H__ #define BLOCKED_INUSE_H__ -#include "hashtable.h" -#include "adlist.h" - struct robj; // defined in server.h struct client; // defined in server.h /* Check if client is blocked by blockInuse */ -#define blockInuse_clientBlocked(c) ((c)->flag.blockInuse_blocked) +int blockInuse_clientBlocked(client *c); /* Initialize blockInuse structures, must be called once during server startup. */ void blockInuse_init(void); @@ -58,20 +55,8 @@ int blockInuse_getNumberOfBlockedClients(void); * Block a client on a set of keys. Duplicate keys are allowed and handled. * * To avoid extra copying, this API keeps references to the passed key objects. - * The caller must ensure that the `keys` array and all key objects are - * heap-allocated and remain valid for the duration of the block. - * - * Preconditions: - * - The client must not already have any blocked or unblocked flags set. - * - * Return value: - * - C_ERR if: - * a. nKeys == 0 - * b. any key is not a sds string object - * c. the client is a replica client - * - C_OK otherwise; the client is blocked until all keys are unblocked. */ -int blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]); +void blockInuse_blockClientOnKeys(client *c, int nKeys, robj *keys[]); /* * Unblock clients blocked on the given key. @@ -92,10 +77,9 @@ void blockInuse_unblockClientsOnAllKeys(void); /* * Unlink a client currently blocked by blockInuse. Typically used when - * a client is being freed while still blocked (e.g., due to memory pressure). + * a client is being freed while still blocked (e.g., client-initiated disconnect). * - * This function removes the client from all blockInuse data structures - * and clears its blockInuse blocked flag. + * This function removes the client from all blockInuse data structures. */ void blockInuse_unlinkClient(client *c); diff --git a/src/networking.c b/src/networking.c index 229ed03537f..9d54ffc9dcb 100644 --- a/src/networking.c +++ b/src/networking.c @@ -40,6 +40,7 @@ #include "module.h" #include "connection.h" #include "zmalloc.h" +#include "blocked_inuse.h" #include #include #include @@ -1902,7 +1903,7 @@ void unlinkClient(client *c) { waitForClientIO(c); /* If this is marked as current client unset it. */ - if (c->conn && server.current_client == c) server.current_client = NULL; + if (server.current_client == c) server.current_client = NULL; if (blockInuse_clientBlocked(c)) blockInuse_unlinkClient(c); @@ -1990,7 +1991,7 @@ void unlinkClient(client *c) { /* Clear the tracking status. */ if (c->flag.tracking) disableTracking(c); - // Here client should never in unblocked or blockInuse blocked state. + // Client should never in unblocked or blockInuse blocked state. serverAssert(!(blockInuse_clientBlocked(c) || (c)->flag.unblocked)); } @@ -4281,7 +4282,7 @@ sds catClientInfoString(sds s, client *client, int hide_user_data) { if (client->flag.pubsub) *p++ = 'P'; if (client->flag.multi) *p++ = 'x'; if (client->flag.blocked) *p++ = 'b'; - if (client->flag.blockInuse_blocked) *p++ = 'X'; + if (blockInuse_clientBlocked(client)) *p++ = 'X'; if (client->flag.tracking) *p++ = 't'; if (client->flag.tracking_broken_redir) *p++ = 'R'; if (client->flag.tracking_bcast) *p++ = 'B'; diff --git a/src/server.c b/src/server.c index e35365a10c4..2e55e61a74d 100644 --- a/src/server.c +++ b/src/server.c @@ -48,6 +48,7 @@ #include "fmtargs.h" #include "io_threads.h" #include "tls.h" +#include "blocked_inuse.h" #include "sds.h" #include "module.h" #include "scripting_engine.h" @@ -1159,45 +1160,36 @@ void getExpensiveClientsInfo(size_t *in_usage, size_t *out_usage) { } /* - * Check if a blockInuse blocked client connection has been closed from the remote side. + * Check if a TCP client connection has been closed from the remote side. * * Returns: - * 1 if the client has been terminated, - * 0 if still alive. + * true if the client has been terminated, + * false if still alive. */ -static int clientsCronCheckBlockInuseClients(client *c) { - if (!c->conn) return 0; // No connection, cannot check +static bool clientsCronTcpIsClosing(client *c) { + if (!c->conn) return false; // No connection, cannot check // Only TCP or TLS clients are relevant - if (c->conn->type != connectionTypeTcp() && c->conn->type != connectionTypeTls()) return 0; + if (c->conn->type != connectionTypeTcp() && c->conn->type != connectionTypeTls()) return false; - // If neither read nor write handler is installed, client is blocked. - if (aeGetFileEvents(server.el, c->conn->fd) != AE_NONE) return 0; + // Skip if event handlers are installed + if (aeGetFileEvents(server.el, c->conn->fd) != AE_NONE) return false; - // No event handler exists, check TCP socket state. - - /* TCP state introspection is platform-specific: - * - Linux: TCP_INFO / struct tcp_info - * - macOS: TCP_CONNECTION_INFO / struct tcp_connection_info (BSD TCP FSM) - */ #if defined(__linux__) + // Check TCP socket state using Linux TCP_INFO struct tcp_info info; socklen_t infolen = sizeof(info); - if (getsockopt(c->conn->fd, IPPROTO_TCP, TCP_INFO, &info, &infolen) != 0 || infolen < sizeof(info)) return 0; // Cannot retrieve TCP info - - if (info.tcpi_state == TCP_CLOSE_WAIT || - info.tcpi_state == TCP_CLOSE) + if (getsockopt(c->conn->fd, IPPROTO_TCP, TCP_INFO, &info, &infolen) != 0 || infolen < sizeof(info)) return false; // Cannot retrieve TCP info + bool connection_is_closing = (info.tcpi_state == TCP_CLOSE_WAIT || info.tcpi_state == TCP_CLOSE); #elif defined(__APPLE__) + // Check TCP socket state using macOS TCP_CONNECTION_INFO struct tcp_connection_info info; socklen_t infolen = sizeof(info); - - if (getsockopt(c->conn->fd, IPPROTO_TCP, TCP_CONNECTION_INFO, &info, &infolen) != 0 || infolen < sizeof(info)) return 0; // Cannot retrieve TCP info - - // Check if connection is closed or half-closed - if (info.tcpi_state == TCPS_CLOSE_WAIT || - info.tcpi_state == TCPS_CLOSED) + if (getsockopt(c->conn->fd, IPPROTO_TCP, TCP_CONNECTION_INFO, &info, &infolen) != 0 || infolen < sizeof(info)) return false; // Cannot retrieve TCP info + bool connection_is_closing = (info.tcpi_state == TCPS_CLOSE_WAIT || info.tcpi_state == TCPS_CLOSED); #endif - { + + if (connection_is_closing) { if (server.verbosity <= LL_VERBOSE) { sds client_info = catClientInfoString(sdsempty(), c, server.hide_user_data_from_log); serverLog(LL_VERBOSE, "Client closed connection while blocked %s", client_info); @@ -1205,10 +1197,10 @@ static int clientsCronCheckBlockInuseClients(client *c) { } freeClientAsync(c); - return 1; // Client has been closed + return true; // Client has been closed } - return 0; // Client is still alive + return false; // Client is still alive } /* This function is called by clientsTimeProc() and is used in order to perform @@ -1265,7 +1257,7 @@ static void clientsCron(int clients_this_cycle) { if (clientsCronResizeQueryBuffer(c)) continue; if (clientsCronResizeOutputBuffer(c, now)) continue; if (clientsCronTrackExpensiveClients(c, curr_peak_mem_usage_slot)) continue; - if (clientsCronCheckBlockInuseClients(c)) continue; + if (clientsCronTcpIsClosing(c)) continue; /* Iterating all the clients in getMemoryOverheadData() is too slow and * in turn would make the INFO command too slow. So we perform this @@ -2992,9 +2984,6 @@ void initServer(void) { server.debug_client_enforce_reply_list = 0; resetReplicationBuffer(); - /* Init blockInuse data structures */ - blockInuse_init(); - /* Make sure the locale is set on startup based on the config file. */ if (setlocale(LC_COLLATE, server.locale_collate) == NULL) { if (server.locale_collate[0] == '\0') { @@ -3140,6 +3129,7 @@ void initServer(void) { commandlogInit(); latencyMonitorInit(); + blockInuse_init(); initSharedQueryBuf(); /* Initialize ACL default password if it exists */ @@ -4277,7 +4267,7 @@ void unprepareCommand(client *c) { * other operations can be performed by the caller. Otherwise * if C_ERR is returned the client was destroyed (i.e. after QUIT). */ int processCommand(client *c) { - serverAssert(!(blockInuse_clientBlocked(c) || c->flag.unblocked == 1)); + serverAssert(!(blockInuse_clientBlocked(c) || c->flag.unblocked == 1 || c->flag.blocked == 1)); if (!scriptIsTimedout()) { /* Both EXEC and scripts call call() directly so there should be @@ -4927,7 +4917,6 @@ int finishShutdown(void) { /* Close the listening sockets. Apparently this allows faster restarts. */ closeListeningSockets(1); - /* Release blockInuse data structures. */ blockInuse_release(); moduleUnloadAllModules(); diff --git a/src/server.h b/src/server.h index 28b2ef18382..1b827e2cae9 100644 --- a/src/server.h +++ b/src/server.h @@ -83,7 +83,6 @@ #include "trace/trace.h" #include "entry.h" #include "lrulfu.h" -#include "blocked_inuse.h" /* * Sanity check: we require large-file support. If include order caused diff --git a/src/timeout.c b/src/timeout.c index 6d834e8cf36..b6522991978 100644 --- a/src/timeout.c +++ b/src/timeout.c @@ -28,6 +28,7 @@ #include "server.h" #include "cluster.h" +#include "blocked_inuse.h" #include @@ -55,11 +56,11 @@ int clientsCronHandleTimeout(client *c, mstime_t now_ms) { if (server.maxidletime && /* This handles the idle clients connection timeout if set. */ - !c->flag.replica && /* No timeout for replicas and monitors */ - !mustObeyClient(c) && /* No timeout for primaries and AOF */ - !c->flag.blocked && /* No timeout for BLPOP */ - !c->flag.pubsub && /* No timeout for Pub/Sub clients */ - !c->flag.blockInuse_blocked && /* No timeout for BlockInuse client */ + !c->flag.replica && /* No timeout for replicas and monitors */ + !mustObeyClient(c) && /* No timeout for primaries and AOF */ + !c->flag.blocked && /* No timeout for BLPOP */ + !c->flag.pubsub && /* No timeout for Pub/Sub clients */ + !blockInuse_clientBlocked(c) && /* No timeout for BlockInuse client */ (now - c->last_interaction > server.maxidletime)) { serverLog(LL_VERBOSE, "Closing idle client"); freeClient(c);