Unverified Commit 6790d848 authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Reuse temporary client objects for blocked clients by module (#9940)

Added a pool for temporary client objects to reuse in module operations.
By reusing temporary clients, we are avoiding expensive createClient()/freeClient()
calls and improving performance of RM_BlockClient() and  RM_GetThreadSafeContext() calls. 

This commit contains two optimizations: 

1 - RM_BlockClient() and RM_GetThreadSafeContext() calls create temporary clients and they are freed in
RM_UnblockClient() and RM_FreeThreadSafeContext() calls respectively. Creating/destroying client object
takes quite time. To avoid that, added a pool of temporary clients. Pool expands when more clients are needed.
Also, added a cron function to shrink the pool and free unused clients after some time. Pool starts with zero
clients in it. It does not have max size and can grow unbounded as we need it. We will keep minimum of 8
temporary clients in the pool once created. Keeping small amount of clients to avoid client allocation costs
if temporary clients are required after some idle period.

2 - After unblocking a client (RM_UnblockClient()), one byte is written to pipe to wake up Redis main thread.
If there are many clients that will be unblocked, each operation requires one write() call which is quite expensive.
Changed code to avoid subsequent calls if possible. 

There are a few more places that need temporary client objects (e.g RM_Call()). These are now using the same
temporary client pool to make things more centralized. 
parent 3204a035
This diff is collapsed.
......@@ -1374,6 +1374,45 @@ void unlinkClient(client *c) {
if (c->flags & CLIENT_TRACKING) disableTracking(c);
}
/* Clear the client state to resemble a newly connected client. */
void clearClientConnectionState(client *c) {
listNode *ln;
/* MONITOR clients are also marked with CLIENT_SLAVE, we need to
* distinguish between the two.
*/
if (c->flags & CLIENT_MONITOR) {
ln = listSearchKey(server.monitors,c);
serverAssert(ln != NULL);
listDelNode(server.monitors,ln);
c->flags &= ~(CLIENT_MONITOR|CLIENT_SLAVE);
}
serverAssert(!(c->flags &(CLIENT_SLAVE|CLIENT_MASTER)));
if (c->flags & CLIENT_TRACKING) disableTracking(c);
selectDb(c,0);
c->resp = 2;
clientSetDefaultAuth(c);
moduleNotifyUserChanged(c);
discardTransaction(c);
pubsubUnsubscribeAllChannels(c,0);
pubsubUnsubscribeShardAllChannels(c, 0);
pubsubUnsubscribeAllPatterns(c,0);
if (c->name) {
decrRefCount(c->name);
c->name = NULL;
}
/* Selectively clear state flags not covered above */
c->flags &= ~(CLIENT_ASKING|CLIENT_READONLY|CLIENT_PUBSUB|
CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP_NEXT);
}
void freeClient(client *c) {
listNode *ln;
......@@ -2598,45 +2637,18 @@ int clientSetNameOrReply(client *c, robj *name) {
/* Reset the client state to resemble a newly connected client.
*/
void resetCommand(client *c) {
listNode *ln;
/* MONITOR clients are also marked with CLIENT_SLAVE, we need to
* distinguish between the two.
*/
if (c->flags & CLIENT_MONITOR) {
ln = listSearchKey(server.monitors,c);
serverAssert(ln != NULL);
listDelNode(server.monitors,ln);
c->flags &= ~(CLIENT_MONITOR|CLIENT_SLAVE);
}
uint64_t flags = c->flags;
if (flags & CLIENT_MONITOR) flags &= ~(CLIENT_MONITOR|CLIENT_SLAVE);
if (c->flags & (CLIENT_SLAVE|CLIENT_MASTER|CLIENT_MODULE)) {
if (flags & (CLIENT_SLAVE|CLIENT_MASTER|CLIENT_MODULE)) {
addReplyError(c,"can only reset normal client connections");
return;
}
if (c->flags & CLIENT_TRACKING) disableTracking(c);
selectDb(c,0);
c->resp = 2;
clientSetDefaultAuth(c);
moduleNotifyUserChanged(c);
discardTransaction(c);
pubsubUnsubscribeAllChannels(c,0);
pubsubUnsubscribeShardAllChannels(c, 0);
pubsubUnsubscribeAllPatterns(c,0);
if (c->name) {
decrRefCount(c->name);
c->name = NULL;
}
/* Selectively clear state flags not covered above */
c->flags &= ~(CLIENT_ASKING|CLIENT_READONLY|CLIENT_PUBSUB|
CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP_NEXT);
clearClientConnectionState(c);
addReplyStatus(c,"RESET");
}
......
......@@ -1320,6 +1320,10 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
server.rdb_bgsave_scheduled = 0;
}
run_with_period(100) {
if (moduleCount()) modulesCron();
}
/* Fire the cron loop modules event. */
RedisModuleCronLoopV1 ei = {REDISMODULE_CRON_LOOP_VERSION,server.hz};
moduleFireServerEvent(REDISMODULE_EVENT_CRON_LOOP,
......
......@@ -1453,7 +1453,6 @@ struct redisServer {
to be processed. */
pid_t child_pid; /* PID of current child */
int child_type; /* Type of current child */
client *module_client; /* "Fake" client to call Redis from modules */
/* Networking */
int port; /* TCP listening port */
int tls_port; /* TLS listening port */
......@@ -2302,6 +2301,7 @@ void populateCommandLegacyRangeSpec(struct redisCommand *c);
/* Modules */
void moduleInitModulesSystem(void);
void moduleInitModulesSystemLast(void);
void modulesCron(void);
int moduleLoad(const char *path, void **argv, int argc);
void moduleLoadFromQueue(void);
int moduleGetCommandKeysViaAPI(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
......@@ -2361,6 +2361,7 @@ void freeClient(client *c);
void freeClientAsync(client *c);
void logInvalidUseAndFreeClientAsync(client *c, const char *fmt, ...);
int beforeNextClient(client *c);
void clearClientConnectionState(client *c);
void resetClient(client *c);
void freeClientOriginalArgv(client *c);
void sendReplyToClient(connection *conn);
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment