-
Notifications
You must be signed in to change notification settings - Fork 944
Added cache for COMMAND #2839
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: unstable
Are you sure you want to change the base?
Added cache for COMMAND #2839
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2342,6 +2342,8 @@ void initServerConfig(void) { | |
| * valkey.conf using the rename-command directive. */ | ||
| server.commands = hashtableCreate(&commandSetType); | ||
| server.orig_commands = hashtableCreate(&originalCommandSetType); | ||
| server.command_response_cache_resp2 = NULL; | ||
| server.command_response_cache_resp3 = NULL; | ||
| populateCommandTable(); | ||
|
|
||
| /* Debugging */ | ||
|
|
@@ -3282,6 +3284,10 @@ int populateCommandStructure(struct serverCommand *c) { | |
| * has been issued for the first time */ | ||
| c->latency_histogram = NULL; | ||
|
|
||
| /* Initialize command info cache */ | ||
| c->info_cache_resp2 = NULL; | ||
| c->info_cache_resp3 = NULL; | ||
|
|
||
| /* Handle the legacy range spec and the "movablekeys" flag (must be done after populating all key specs). */ | ||
| populateCommandLegacyRangeSpec(c); | ||
|
|
||
|
|
@@ -5222,30 +5228,78 @@ void addReplyCommandSubCommands(client *c, | |
| hashtableResetIterator(&iter); | ||
| } | ||
|
|
||
| /* Collect all output from a caching client (both buffer and reply list) */ | ||
| static sds collectCachedResponse(client *c) { | ||
| sds response = sdsempty(); | ||
|
|
||
| /* First, collect from the fixed buffer if any */ | ||
| if (c->bufpos > 0) { | ||
| response = sdscatlen(response, c->buf, c->bufpos); | ||
| } | ||
|
|
||
| /* Then, collect from the reply list */ | ||
| listIter li; | ||
| listNode *ln; | ||
| clientReplyBlock *val_block; | ||
| listRewind(c->reply, &li); | ||
| while ((ln = listNext(&li)) != NULL) { | ||
| val_block = (clientReplyBlock *)listNodeValue(ln); | ||
| response = sdscatlen(response, val_block->buf, val_block->used); | ||
| } | ||
|
|
||
| return response; | ||
| } | ||
|
|
||
| /* Forward declaration */ | ||
| void addReplyCommandInfo(client *c, struct serverCommand *cmd); | ||
|
Comment on lines
+5253
to
+5254
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Just move
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. there is a recursion to process subcommands |
||
|
|
||
| /* Generate and cache the command info response for a given protocol version */ | ||
| static void cacheCommandInfo(struct serverCommand *cmd, int resp) { | ||
| client *caching_client = createCachedResponseClient(resp); | ||
|
|
||
| int firstkey = 0, lastkey = 0, keystep = 0; | ||
| if (cmd->legacy_range_key_spec.begin_search_type != KSPEC_BS_INVALID) { | ||
| firstkey = cmd->legacy_range_key_spec.bs.index.pos; | ||
| lastkey = cmd->legacy_range_key_spec.fk.range.lastkey; | ||
| if (lastkey >= 0) lastkey += firstkey; | ||
| keystep = cmd->legacy_range_key_spec.fk.range.keystep; | ||
| } | ||
|
|
||
| addReplyArrayLen(caching_client, 10); | ||
| addReplyBulkCBuffer(caching_client, cmd->fullname, sdslen(cmd->fullname)); | ||
| addReplyLongLong(caching_client, cmd->arity); | ||
| addReplyFlagsForCommand(caching_client, cmd); | ||
| addReplyLongLong(caching_client, firstkey); | ||
| addReplyLongLong(caching_client, lastkey); | ||
| addReplyLongLong(caching_client, keystep); | ||
| addReplyCommandCategories(caching_client, cmd); | ||
| addReplyCommandTips(caching_client, cmd); | ||
| addReplyCommandKeySpecs(caching_client, cmd); | ||
| addReplyCommandSubCommands(caching_client, cmd, addReplyCommandInfo, 0); | ||
|
|
||
| if (resp == 2) { | ||
| cmd->info_cache_resp2 = collectCachedResponse(caching_client); | ||
| } else { | ||
| cmd->info_cache_resp3 = collectCachedResponse(caching_client); | ||
| } | ||
|
|
||
| deleteCachedResponseClient(caching_client); | ||
| } | ||
|
|
||
| /* Output the representation of a server command. Used by the COMMAND command and COMMAND INFO. */ | ||
| void addReplyCommandInfo(client *c, struct serverCommand *cmd) { | ||
| if (!cmd) { | ||
| addReplyNull(c); | ||
| } else { | ||
| int firstkey = 0, lastkey = 0, keystep = 0; | ||
| if (cmd->legacy_range_key_spec.begin_search_type != KSPEC_BS_INVALID) { | ||
| firstkey = cmd->legacy_range_key_spec.bs.index.pos; | ||
| lastkey = cmd->legacy_range_key_spec.fk.range.lastkey; | ||
| if (lastkey >= 0) lastkey += firstkey; | ||
| keystep = cmd->legacy_range_key_spec.fk.range.keystep; | ||
| /* Use cached response if available for the client's protocol version */ | ||
| sds cache = (c->resp == 2) ? cmd->info_cache_resp2 : cmd->info_cache_resp3; | ||
|
|
||
| if (cache == NULL) { | ||
| cacheCommandInfo(cmd, c->resp); | ||
| cache = (c->resp == 2) ? cmd->info_cache_resp2 : cmd->info_cache_resp3; | ||
| } | ||
|
|
||
| addReplyArrayLen(c, 10); | ||
| addReplyBulkCBuffer(c, cmd->fullname, sdslen(cmd->fullname)); | ||
| addReplyLongLong(c, cmd->arity); | ||
| addReplyFlagsForCommand(c, cmd); | ||
| addReplyLongLong(c, firstkey); | ||
| addReplyLongLong(c, lastkey); | ||
| addReplyLongLong(c, keystep); | ||
| addReplyCommandCategories(c, cmd); | ||
| addReplyCommandTips(c, cmd); | ||
| addReplyCommandKeySpecs(c, cmd); | ||
| addReplyCommandSubCommands(c, cmd, addReplyCommandInfo, 0); | ||
|
|
||
| addReplyProto(c, cache, sdslen(cache)); | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -5370,17 +5424,52 @@ void getKeysSubcommand(client *c) { | |
| getKeysSubcommandImpl(c, 0); | ||
| } | ||
|
|
||
| /* COMMAND (no args) */ | ||
| void commandCommand(client *c) { | ||
| /* Invalidate the cached COMMAND response when command table changes */ | ||
| void invalidateCommandCache(void) { | ||
| if (server.command_response_cache_resp2) { | ||
| sdsfree(server.command_response_cache_resp2); | ||
| server.command_response_cache_resp2 = NULL; | ||
| } | ||
| if (server.command_response_cache_resp3) { | ||
| sdsfree(server.command_response_cache_resp3); | ||
| server.command_response_cache_resp3 = NULL; | ||
| } | ||
| } | ||
|
|
||
| /* Generate and cache the full COMMAND response */ | ||
| static void cacheCommandResponse(int resp) { | ||
| client *caching_client = createCachedResponseClient(resp); | ||
|
|
||
| hashtableIterator iter; | ||
| void *next; | ||
| addReplyArrayLen(c, hashtableSize(server.commands)); | ||
| addReplyArrayLen(caching_client, hashtableSize(server.commands)); | ||
| hashtableInitIterator(&iter, server.commands, 0); | ||
| while (hashtableNext(&iter, &next)) { | ||
| struct serverCommand *cmd = next; | ||
| addReplyCommandInfo(c, cmd); | ||
| addReplyCommandInfo(caching_client, cmd); | ||
| } | ||
| hashtableResetIterator(&iter); | ||
|
|
||
| if (resp == 2) { | ||
| server.command_response_cache_resp2 = collectCachedResponse(caching_client); | ||
| } else { | ||
| server.command_response_cache_resp3 = collectCachedResponse(caching_client); | ||
| } | ||
|
|
||
| deleteCachedResponseClient(caching_client); | ||
| } | ||
|
|
||
| /* COMMAND (no args) */ | ||
| void commandCommand(client *c) { | ||
| /* Use cached response if available for the client's protocol version */ | ||
| sds cache = (c->resp == 2) ? server.command_response_cache_resp2 : server.command_response_cache_resp3; | ||
|
|
||
| if (cache == NULL) { | ||
| cacheCommandResponse(c->resp); | ||
| cache = (c->resp == 2) ? server.command_response_cache_resp2 : server.command_response_cache_resp3; | ||
| } | ||
|
|
||
| addReplyProto(c, cache, sdslen(cache)); | ||
| } | ||
|
|
||
| /* COMMAND COUNT */ | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -1677,6 +1677,8 @@ struct valkeyServer { | |
| serverDb **db; /* each db created when it's first used */ | ||
| hashtable *commands; /* Command table */ | ||
| hashtable *orig_commands; /* Command table before command renaming. */ | ||
| sds command_response_cache_resp2; /* Cached COMMAND response for RESP2 */ | ||
| sds command_response_cache_resp3; /* Cached COMMAND response for RESP3 */ | ||
| aeEventLoop *el; | ||
| _Atomic AeIoState io_poll_state; /* Indicates the state of the IO polling. */ | ||
| int io_ae_fired_events; /* Number of poll events received by the IO thread. */ | ||
|
|
@@ -2626,6 +2628,8 @@ struct serverCommand { | |
| * (not the fullname), and the value is the serverCommand structure pointer. */ | ||
| struct serverCommand *parent; | ||
| struct ValkeyModuleCommand *module_cmd; /* A pointer to the module command data (NULL if native command) */ | ||
| sds info_cache_resp2; /* Cached COMMAND INFO response for RESP2 */ | ||
| sds info_cache_resp3; /* Cached COMMAND INFO response for RESP3 */ | ||
|
Comment on lines
+2631
to
+2632
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Would recommend building an array similar to |
||
| }; | ||
|
|
||
| struct serverError { | ||
|
|
@@ -3783,6 +3787,7 @@ void commandCommand(client *c); | |
| void commandCountCommand(client *c); | ||
| void commandListCommand(client *c); | ||
| void commandInfoCommand(client *c); | ||
| void invalidateCommandCache(void); | ||
| void commandGetKeysCommand(client *c); | ||
| void commandGetKeysAndFlagsCommand(client *c); | ||
| void commandHelpCommand(client *c); | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This should be a common function associated with a cached response client. It's unclear why this is different from
aggregateClientOutputBufferand I tagged @roshkhatri for info.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
aggregateClientOutputBuffer assumes c->bufpos == 0 which is not the case here as Add* functions utilize c->buf
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
FYI - I talked with @roshkhatri. He agrees that the existing function
aggregateClientOutputBuffershould be updated rather than creating a new function. His existing function assumedc->bufpos == 0for his specific use case, but this does not have to be generally true. We can update that function.