Unverified Commit fb4e0d40 authored by Oran Agra's avatar Oran Agra Committed by GitHub
Browse files

Merge pull request #10532 from oranagra/7.0-rc3

Release 7.0 rc3
parents d2b5a579 8b242ef9
......@@ -218,9 +218,12 @@ void dbOverwrite(redisDb *db, robj *key, robj *val) {
val->lru = old->lru;
}
/* Although the key is not really deleted from the database, we regard
overwrite as two steps of unlink+add, so we still need to call the unlink
callback of the module. */
* overwrite as two steps of unlink+add, so we still need to call the unlink
* callback of the module. */
moduleNotifyKeyUnlink(key,old,db->id);
/* We want to try to unblock any client using a blocking XREADGROUP */
if (old->type == OBJ_STREAM)
signalKeyAsReady(db,key,old->type);
dictSetVal(db->dict, de, val);
if (server.lazyfree_lazy_server_del) {
......@@ -311,6 +314,9 @@ static int dbGenericDelete(redisDb *db, robj *key, int async) {
robj *val = dictGetVal(de);
/* Tells the module that the key has been unlinked from the database. */
moduleNotifyKeyUnlink(key,val,db->id);
/* We want to try to unblock any client using a blocking XREADGROUP */
if (val->type == OBJ_STREAM)
signalKeyAsReady(db,key,val->type);
if (async) {
freeObjAsync(key, val, db->id);
dictSetVal(db->dict, de, NULL);
......@@ -551,6 +557,7 @@ void signalFlushedDb(int dbid, int async) {
}
for (int j = startdb; j <= enddb; j++) {
scanDatabaseForDeletedStreams(&server.db[j], NULL);
touchAllWatchedKeysInDb(&server.db[j], NULL);
}
......@@ -1311,7 +1318,7 @@ void copyCommand(client *c) {
* one or more blocked clients for B[LR]POP or other blocking commands
* and signal the keys as ready if they are of the right type. See the comment
* where the function is used for more info. */
void scanDatabaseForReadyLists(redisDb *db) {
void scanDatabaseForReadyKeys(redisDb *db) {
dictEntry *de;
dictIterator *di = dictGetSafeIterator(db->blocking_keys);
while((de = dictNext(di)) != NULL) {
......@@ -1325,6 +1332,39 @@ void scanDatabaseForReadyLists(redisDb *db) {
dictReleaseIterator(di);
}
/* Since we are unblocking XREADGROUP clients in the event the
* key was deleted/overwritten we must do the same in case the
* database was flushed/swapped. */
void scanDatabaseForDeletedStreams(redisDb *emptied, redisDb *replaced_with) {
/* Optimization: If no clients are in type BLOCKED_STREAM,
* we can skip this loop. */
if (!server.blocked_clients_by_type[BLOCKED_STREAM]) return;
dictEntry *de;
dictIterator *di = dictGetSafeIterator(emptied->blocking_keys);
while((de = dictNext(di)) != NULL) {
robj *key = dictGetKey(de);
int was_stream = 0, is_stream = 0;
dictEntry *kde = dictFind(emptied->dict, key->ptr);
if (kde) {
robj *value = dictGetVal(kde);
was_stream = value->type == OBJ_STREAM;
}
if (replaced_with) {
dictEntry *kde = dictFind(replaced_with->dict, key->ptr);
if (kde) {
robj *value = dictGetVal(kde);
is_stream = value->type == OBJ_STREAM;
}
}
/* We want to try to unblock any client using a blocking XREADGROUP */
if (was_stream && !is_stream)
signalKeyAsReady(emptied, key, OBJ_STREAM);
}
dictReleaseIterator(di);
}
/* Swap two databases at runtime so that all clients will magically see
* the new database even if already connected. Note that the client
* structure c->db points to a given DB, so we need to be smarter and
......@@ -1345,6 +1385,10 @@ int dbSwapDatabases(int id1, int id2) {
touchAllWatchedKeysInDb(db1, db2);
touchAllWatchedKeysInDb(db2, db1);
/* Try to unblock any XREADGROUP clients if the key no longer exists. */
scanDatabaseForDeletedStreams(db1, db2);
scanDatabaseForDeletedStreams(db2, db1);
/* Swap hash tables. Note that we don't swap blocking_keys,
* ready_keys and watched_keys, since we want clients to
* remain in the same DB they were. */
......@@ -1367,8 +1411,8 @@ int dbSwapDatabases(int id1, int id2) {
* in dbAdd() when a list is created. So here we need to rescan
* the list of clients blocked on lists and signal lists as ready
* if needed. */
scanDatabaseForReadyLists(db1);
scanDatabaseForReadyLists(db2);
scanDatabaseForReadyKeys(db1);
scanDatabaseForReadyKeys(db2);
return C_OK;
}
......@@ -1391,6 +1435,9 @@ void swapMainDbWithTempDb(redisDb *tempDb) {
* client watching keys. */
touchAllWatchedKeysInDb(activedb, newdb);
/* Try to unblock any XREADGROUP clients if the key no longer exists. */
scanDatabaseForDeletedStreams(activedb, newdb);
/* Swap hash tables. Note that we don't swap blocking_keys,
* ready_keys and watched_keys, since clients
* remain in the same DB they were. */
......@@ -1413,7 +1460,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) {
* in dbAdd() when a list is created. So here we need to rescan
* the list of clients blocked on lists and signal lists as ready
* if needed. */
scanDatabaseForReadyLists(activedb);
scanDatabaseForReadyKeys(activedb);
}
trackingInvalidateKeysOnFlush(1);
......
......@@ -482,10 +482,12 @@ void debugCommand(client *c) {
" Show low level client eviction pools info (maxmemory-clients).",
"PAUSE-CRON <0|1>",
" Stop periodic cron job processing.",
"REPLYBUFFER-PEAK-RESET-TIME <NEVER||RESET|time>",
"REPLYBUFFER PEAK-RESET-TIME <NEVER||RESET|time>",
" Sets the time (in milliseconds) to wait between client reply buffer peak resets.",
" In case NEVER is provided the last observed peak will never be reset",
" In case RESET is provided the peak reset time will be restored to the default value",
"REPLYBUFFER RESIZING <0|1>",
" Enable or disable the replay buffer resize cron job",
NULL
};
addReplyHelp(c, help);
......@@ -793,6 +795,10 @@ NULL
* also have a normal reply type after the attribute. */
addReplyBulkCString(c,"Some real reply following the attribute");
} else if (!strcasecmp(name,"push")) {
if (c->resp < 3) {
addReplyError(c,"RESP2 is not supported by this command");
return;
}
addReplyPushLen(c,2);
addReplyBulkCString(c,"server-cpu-usage");
addReplyLongLong(c,42);
......@@ -962,14 +968,21 @@ NULL
{
server.pause_cron = atoi(c->argv[2]->ptr);
addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"replybuffer-peak-reset-time") && c->argc == 3 ) {
if (!strcasecmp(c->argv[2]->ptr, "never")) {
server.reply_buffer_peak_reset_time = -1;
} else if(!strcasecmp(c->argv[2]->ptr, "reset")) {
server.reply_buffer_peak_reset_time = REPLY_BUFFER_DEFAULT_PEAK_RESET_TIME;
} else if (!strcasecmp(c->argv[1]->ptr,"replybuffer") && c->argc == 4 ) {
if(!strcasecmp(c->argv[2]->ptr, "peak-reset-time")) {
if (!strcasecmp(c->argv[3]->ptr, "never")) {
server.reply_buffer_peak_reset_time = -1;
} else if(!strcasecmp(c->argv[3]->ptr, "reset")) {
server.reply_buffer_peak_reset_time = REPLY_BUFFER_DEFAULT_PEAK_RESET_TIME;
} else {
if (getLongFromObjectOrReply(c, c->argv[3], &server.reply_buffer_peak_reset_time, NULL) != C_OK)
return;
}
} else if(!strcasecmp(c->argv[2]->ptr,"resizing")) {
server.reply_buffer_resizing_enabled = atoi(c->argv[3]->ptr);
} else {
if (getLongFromObjectOrReply(c, c->argv[2], &server.reply_buffer_peak_reset_time, NULL) != C_OK)
return;
addReplySubcommandSyntaxError(c);
return;
}
addReply(c, shared.ok);
} else {
......@@ -1943,7 +1956,7 @@ void sigsegvHandler(int sig, siginfo_t *info, void *secret) {
serverLog(LL_WARNING,
"Accessing address: %p", (void*)info->si_addr);
}
if (info->si_code <= SI_USER && info->si_pid != -1) {
if (info->si_code == SI_USER && info->si_pid != -1) {
serverLog(LL_WARNING, "Killed by PID: %ld, UID: %d", (long) info->si_pid, info->si_uid);
}
......
......@@ -407,7 +407,7 @@ long activeDefragSdsListAndDict(list *l, dict *d, int dict_val_type) {
* new pointer. Additionally, we try to defrag the dictEntry in that dict.
* Oldkey mey be a dead pointer and should not be accessed (we get a
* pre-calculated hash value). Newkey may be null if the key pointer wasn't
* moved. Return value is the the dictEntry if found, or NULL if not found.
* moved. Return value is the dictEntry if found, or NULL if not found.
* NOTE: this is very ugly code, but it let's us avoid the complication of
* doing a scan on another dict. */
dictEntry* replaceSatelliteDictKeyPtrAndOrDefragDictEntry(dict *d, sds oldkey, sds newkey, uint64_t hash, long *defragged) {
......@@ -1196,7 +1196,7 @@ void activeDefragCycle(void) {
cursor = dictScan(db->dict, cursor, defragScanCallback, defragDictBucketCallback, db);
/* Once in 16 scan iterations, 512 pointer reallocations. or 64 keys
* (if we have a lot of pointers in one hash bucket or rehasing),
* (if we have a lot of pointers in one hash bucket or rehashing),
* check if we reached the time limit.
* But regardless, don't start a new db in this loop, this is because after
* the last db we call defragOtherGlobals, which must be done in one cycle */
......
......@@ -508,6 +508,7 @@ void evalGenericCommand(client *c, int evalsha) {
scriptRunCtx rctx;
if (scriptPrepareForRun(&rctx, lctx.lua_client, c, lua_cur_script, l->flags, ro) != C_OK) {
lua_pop(lua,2); /* Remove the function and error handler. */
return;
}
rctx.flags |= SCRIPT_EVAL_MODE; /* mark the current run as EVAL (as opposed to FCALL) so we'll
......
......@@ -57,6 +57,12 @@ struct functionsLibCtx {
dict *engines_stats; /* Per engine statistics */
};
typedef struct functionsLibMataData {
sds engine;
sds name;
sds code;
} functionsLibMataData;
dictType engineDictType = {
dictSdsCaseHash, /* hash function */
dictSdsDup, /* key dup */
......@@ -124,7 +130,6 @@ static size_t functionMallocSize(functionInfo *fi) {
static size_t libraryMallocSize(functionLibInfo *li) {
return zmalloc_size(li) + sdsZmallocSize(li->name)
+ (li->desc ? sdsZmallocSize(li->desc) : 0)
+ sdsZmallocSize(li->code);
}
......@@ -157,7 +162,6 @@ static void engineLibraryFree(functionLibInfo* li) {
dictRelease(li->functions);
sdsfree(li->name);
sdsfree(li->code);
if (li->desc) sdsfree(li->desc);
zfree(li);
}
......@@ -265,14 +269,13 @@ int functionLibCreateFunction(sds name, void *function, functionLibInfo *li, sds
return C_OK;
}
static functionLibInfo* engineLibraryCreate(sds name, engineInfo *ei, sds desc, sds code) {
static functionLibInfo* engineLibraryCreate(sds name, engineInfo *ei, sds code) {
functionLibInfo *li = zmalloc(sizeof(*li));
*li = (functionLibInfo) {
.name = sdsdup(name),
.functions = dictCreate(&libraryFunctionDictType),
.ei = ei,
.code = sdsdup(code),
.desc = desc ? sdsdup(desc) : NULL,
};
return li;
}
......@@ -540,17 +543,11 @@ void functionListCommand(client *c) {
}
}
++reply_len;
addReplyMapLen(c, with_code? 5 : 4);
addReplyMapLen(c, with_code? 4 : 3);
addReplyBulkCString(c, "library_name");
addReplyBulkCBuffer(c, li->name, sdslen(li->name));
addReplyBulkCString(c, "engine");
addReplyBulkCBuffer(c, li->ei->name, sdslen(li->ei->name));
addReplyBulkCString(c, "description");
if (li->desc) {
addReplyBulkCBuffer(c, li->desc, sdslen(li->desc));
} else {
addReplyNull(c);
}
addReplyBulkCString(c, "functions");
addReplyArrayLen(c, dictSize(li->functions));
......@@ -745,11 +742,11 @@ void functionRestoreCommand(client *c) {
err = sdsnew("can not read data type");
goto load_error;
}
if (type != RDB_OPCODE_FUNCTION) {
if (type != RDB_OPCODE_FUNCTION && type != RDB_OPCODE_FUNCTION2) {
err = sdsnew("given type is not a function");
goto load_error;
}
if (rdbFunctionLoad(&payload, rdbver, functions_lib_ctx, RDBFLAGS_NONE, &err) != C_OK) {
if (rdbFunctionLoad(&payload, rdbver, functions_lib_ctx, type, RDBFLAGS_NONE, &err) != C_OK) {
if (!err) {
err = sdsnew("failed loading the given functions payload");
}
......@@ -868,36 +865,111 @@ static int functionsVerifyName(sds name) {
return C_OK;
}
/* Compile and save the given library, return C_OK on success and C_ERR on failure.
* In case on failure the err out param is set with relevant error message */
int functionsCreateWithLibraryCtx(sds lib_name,sds engine_name, sds desc, sds code,
int replace, sds* err, functionsLibCtx *lib_ctx) {
int functionExtractLibMetaData(sds payload, functionsLibMataData *md, sds *err) {
sds name = NULL;
sds desc = NULL;
sds engine = NULL;
sds code = NULL;
if (strncmp(payload, "#!", 2) != 0) {
*err = sdsnew("Missing library metadata");
return C_ERR;
}
char *shebang_end = strchr(payload, '\n');
if (shebang_end == NULL) {
*err = sdsnew("Invalid library metadata");
return C_ERR;
}
size_t shebang_len = shebang_end - payload;
sds shebang = sdsnewlen(payload, shebang_len);
int numparts;
sds *parts = sdssplitargs(shebang, &numparts);
sdsfree(shebang);
if (!parts || numparts == 0) {
*err = sdsnew("Invalid library metadata");
sdsfreesplitres(parts, numparts);
return C_ERR;
}
engine = sdsdup(parts[0]);
sdsrange(engine, 2, -1);
for (int i = 1 ; i < numparts ; ++i) {
sds part = parts[i];
if (strncasecmp(part, "name=", 5) == 0) {
if (name) {
*err = sdscatfmt(sdsempty(), "Invalid metadata value, name argument was given multiple times");
goto error;
}
name = sdsdup(part);
sdsrange(name, 5, -1);
continue;
}
*err = sdscatfmt(sdsempty(), "Invalid metadata value given: %s", part);
goto error;
}
if (!name) {
*err = sdsnew("Library name was not given");
goto error;
}
sdsfreesplitres(parts, numparts);
md->name = name;
md->code = sdsnewlen(shebang_end, sdslen(payload) - shebang_len);
md->engine = engine;
return C_OK;
error:
if (name) sdsfree(name);
if (desc) sdsfree(desc);
if (engine) sdsfree(engine);
if (code) sdsfree(code);
sdsfreesplitres(parts, numparts);
return C_ERR;
}
void functionFreeLibMetaData(functionsLibMataData *md) {
if (md->code) sdsfree(md->code);
if (md->name) sdsfree(md->name);
if (md->engine) sdsfree(md->engine);
}
/* Compile and save the given library, return the loaded library name on success
* and NULL on failure. In case on failure the err out param is set with relevant error message */
sds functionsCreateWithLibraryCtx(sds code, int replace, sds* err, functionsLibCtx *lib_ctx) {
dictIterator *iter = NULL;
dictEntry *entry = NULL;
if (functionsVerifyName(lib_name)) {
functionLibInfo *new_li = NULL;
functionLibInfo *old_li = NULL;
functionsLibMataData md = {0};
if (functionExtractLibMetaData(code, &md, err) != C_OK) {
return NULL;
}
if (functionsVerifyName(md.name)) {
*err = sdsnew("Library names can only contain letters and numbers and must be at least one character long");
return C_ERR;
goto error;
}
engineInfo *ei = dictFetchValue(engines, engine_name);
engineInfo *ei = dictFetchValue(engines, md.engine);
if (!ei) {
*err = sdsnew("Engine not found");
return C_ERR;
*err = sdscatfmt(sdsempty(), "Engine '%S' not found", md.engine);
goto error;
}
engine *engine = ei->engine;
functionLibInfo *old_li = dictFetchValue(lib_ctx->libraries, lib_name);
old_li = dictFetchValue(lib_ctx->libraries, md.name);
if (old_li && !replace) {
*err = sdsnew("Library already exists");
return C_ERR;
*err = sdscatfmt(sdsempty(), "Library '%S' already exists", md.name);
goto error;
}
if (old_li) {
libraryUnlink(lib_ctx, old_li);
}
functionLibInfo *new_li = engineLibraryCreate(lib_name, ei, desc, code);
if (engine->create(engine->engine_ctx, new_li, code, err) != C_OK) {
new_li = engineLibraryCreate(md.name, ei, code);
if (engine->create(engine->engine_ctx, new_li, md.code, err) != C_OK) {
goto error;
}
......@@ -925,48 +997,34 @@ int functionsCreateWithLibraryCtx(sds lib_name,sds engine_name, sds desc, sds co
engineLibraryFree(old_li);
}
return C_OK;
sds loaded_lib_name = md.name;
md.name = NULL;
functionFreeLibMetaData(&md);
return loaded_lib_name;
error:
if (iter) dictReleaseIterator(iter);
engineLibraryFree(new_li);
if (old_li) {
libraryLink(lib_ctx, old_li);
}
return C_ERR;
if (new_li) engineLibraryFree(new_li);
if (old_li) libraryLink(lib_ctx, old_li);
functionFreeLibMetaData(&md);
return NULL;
}
/*
* FUNCTION LOAD <ENGINE NAME> <LIBRARY NAME>
* [REPLACE] [DESC <LIBRARY DESCRIPTION>] <LIBRARY CODE>
*
* ENGINE NAME - name of the engine to use the run the library
* LIBRARY NAME - name of the library
* FUNCTION LOAD [REPLACE] <LIBRARY CODE>
* REPLACE - optional, replace existing library
* DESCRIPTION - optional, library description
* LIBRARY CODE - library code to pass to the engine
*/
void functionLoadCommand(client *c) {
robj *engine_name = c->argv[2];
robj *library_name = c->argv[3];
int replace = 0;
int argc_pos = 4;
sds desc = NULL;
int argc_pos = 2;
while (argc_pos < c->argc - 1) {
robj *next_arg = c->argv[argc_pos++];
if (!strcasecmp(next_arg->ptr, "replace")) {
replace = 1;
continue;
}
if (!strcasecmp(next_arg->ptr, "description")) {
if (argc_pos >= c->argc) {
addReplyError(c, "Bad function description");
return;
}
desc = c->argv[argc_pos++]->ptr;
continue;
}
addReplyErrorFormat(c, "Unknown option given: %s", (char*)next_arg->ptr);
return;
}
......@@ -978,8 +1036,8 @@ void functionLoadCommand(client *c) {
robj *code = c->argv[argc_pos];
sds err = NULL;
if (functionsCreateWithLibraryCtx(library_name->ptr, engine_name->ptr,
desc, code->ptr, replace, &err, curr_functions_lib_ctx) != C_OK)
sds library_name = NULL;
if (!(library_name = functionsCreateWithLibraryCtx(code->ptr, replace, &err, curr_functions_lib_ctx)))
{
addReplyErrorSds(c, err);
return;
......@@ -987,7 +1045,7 @@ void functionLoadCommand(client *c) {
/* Indicate that the command changed the data so it will be replicated and
* counted as a data change (for persistence configuration) */
server.dirty++;
addReply(c, shared.ok);
addReplyBulkSds(c, library_name);
}
/* Return memory usage of all the engines combine */
......
......@@ -106,12 +106,10 @@ struct functionLibInfo {
dict *functions; /* Functions dictionary */
engineInfo *ei; /* Pointer to the function engine */
sds code; /* Library code */
sds desc; /* Library description */
};
int functionsRegisterEngine(const char *engine_name, engine *engine_ctx);
int functionsCreateWithLibraryCtx(sds lib_name, sds engine_name, sds desc, sds code,
int replace, sds* err, functionsLibCtx *lib_ctx);
sds functionsCreateWithLibraryCtx(sds code, int replace, sds* err, functionsLibCtx *lib_ctx);
unsigned long functionsMemory();
unsigned long functionsMemoryOverhead();
unsigned long functionsNum();
......
......@@ -46,7 +46,7 @@
/* Interleave lower bits of x and y, so the bits of x
* are in the even positions and bits from y in the odd;
* x and y must initially be less than 2**32 (65536).
* x and y must initially be less than 2**32 (4294967296).
* From: https://graphics.stanford.edu/~seander/bithacks.html#InterleaveBMN
*/
static inline uint64_t interleave64(uint32_t xlo, uint32_t ylo) {
......
......@@ -399,6 +399,11 @@ struct commandHelp {
"Bind a hash slot to a specific node",
12,
"3.0.0" },
{ "CLUSTER SHARDS",
"",
"Get array of cluster slots to node mappings",
12,
"7.0.0" },
{ "CLUSTER SLAVES",
"node-id",
"List replica nodes of the specified master node",
......@@ -431,7 +436,7 @@ struct commandHelp {
"2.8.13" },
{ "COMMAND GETKEYSANDFLAGS",
"",
"Extract keys given a full Redis command",
"Extract keys and access flags given a full Redis command",
9,
"7.0.0" },
{ "COMMAND HELP",
......@@ -630,7 +635,7 @@ struct commandHelp {
10,
"7.0.0" },
{ "FUNCTION LOAD",
"engine-name library-name [REPLACE] [DESCRIPTION library-description] function-code",
"[REPLACE] function-code",
"Create a function with the given arguments (name, code, description)",
10,
"7.0.0" },
......@@ -1019,6 +1024,11 @@ struct commandHelp {
"Load a module",
9,
"4.0.0" },
{ "MODULE LOADEX",
"path [CONFIG name value [name value ...]] [ARGS arg [arg ...]]",
"Load a module with extended parameters",
9,
"7.0.0" },
{ "MODULE UNLOAD",
"name",
"Unload a module",
......
......@@ -958,7 +958,7 @@ unsigned char *lpPrependInteger(unsigned char *lp, long long lval) {
return lpInsertInteger(lp, lval, p, LP_BEFORE, NULL);
}
/* Append the specified element 'ele' of length 'len' at the end of the
/* Append the specified element 'ele' of length 'size' at the end of the
* listpack. It is implemented in terms of lpInsert(), so the return value is
* the same as lpInsert(). */
unsigned char *lpAppend(unsigned char *lp, unsigned char *ele, uint32_t size) {
......
......@@ -108,7 +108,7 @@ void nolocks_localtime(struct tm *tmp, time_t t, time_t tz, int dst) {
int main(void) {
/* Obtain timezone and daylight info. */
tzset(); /* Now 'timezome' global is populated. */
tzset(); /* Now 'timezone' global is populated. */
time_t t = time(NULL);
struct tm *aux = localtime(&t);
int daylight_active = aux->tm_isdst;
......
This diff is collapsed.
......@@ -31,7 +31,6 @@
* POSSIBILITY OF SUCH DAMAGE.
*/
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h"
#include <pthread.h>
#include <unistd.h>
......
......@@ -31,7 +31,6 @@
* POSSIBILITY OF SUCH DAMAGE.
*/
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h"
#include <stdio.h>
#include <stdlib.h>
......
......@@ -30,7 +30,6 @@
* POSSIBILITY OF SUCH DAMAGE.
*/
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h"
#include <stdio.h>
#include <stdlib.h>
......
......@@ -33,7 +33,6 @@
* POSSIBILITY OF SUCH DAMAGE.
*/
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h"
#include <stdio.h>
#include <stdlib.h>
......
......@@ -30,7 +30,6 @@
* POSSIBILITY OF SUCH DAMAGE.
*/
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h"
#include <stdio.h>
#include <stdlib.h>
......
......@@ -30,7 +30,6 @@
* POSSIBILITY OF SUCH DAMAGE.
*/
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h"
#include <stdio.h>
#include <stdlib.h>
......
......@@ -147,7 +147,6 @@ client *createClient(connection *conn) {
c->ref_block_pos = 0;
c->qb_pos = 0;
c->querybuf = sdsempty();
c->pending_querybuf = sdsempty();
c->querybuf_peak = 0;
c->reqtype = 0;
c->argc = 0;
......@@ -167,6 +166,7 @@ client *createClient(connection *conn) {
c->repl_start_cmd_stream_on_ack = 0;
c->reploff = 0;
c->read_reploff = 0;
c->repl_applied = 0;
c->repl_ack_off = 0;
c->repl_ack_time = 0;
c->repl_last_partial_write = 0;
......@@ -201,7 +201,7 @@ client *createClient(connection *conn) {
c->pending_read_list_node = NULL;
c->client_tracking_redirection = 0;
c->client_tracking_prefixes = NULL;
c->last_memory_usage = c->last_memory_usage_on_bucket_update = 0;
c->last_memory_usage = 0;
c->last_memory_type = CLIENT_TYPE_NORMAL;
c->auth_callback = NULL;
c->auth_callback_privdata = NULL;
......@@ -1568,7 +1568,6 @@ void freeClient(client *c) {
/* Free the query buffer */
sdsfree(c->querybuf);
sdsfree(c->pending_querybuf);
c->querybuf = NULL;
/* Deallocate structures used to block on blocking ops. */
......@@ -1940,7 +1939,7 @@ int writeToClient(client *c, int handler_installed) {
if (!clientHasPendingReplies(c)) {
c->sentlen = 0;
/* Note that writeToClient() is called in a threaded way, but
* adDeleteFileEvent() is not thread safe: however writeToClient()
* aeDeleteFileEvent() is not thread safe: however writeToClient()
* is always called with handler_installed set to 0 from threads
* so we are fine. */
if (handler_installed) {
......@@ -1954,7 +1953,11 @@ int writeToClient(client *c, int handler_installed) {
return C_ERR;
}
}
updateClientMemUsage(c);
/* Update client's memory usage after writing.
* Since this isn't thread safe we do this conditionally. In case of threaded writes this is done in
* handleClientsWithPendingWritesUsingThreads(). */
if (io_threads_op == IO_THREADS_OP_IDLE)
updateClientMemUsage(c);
return C_OK;
}
......@@ -2128,7 +2131,7 @@ int processInlineBuffer(client *c) {
* we got some desynchronization in the protocol, for example
* because of a PSYNC gone bad.
*
* However the is an exception: masters may send us just a newline
* However there is an exception: masters may send us just a newline
* to keep the connection active. */
if (querylen != 0 && c->flags & CLIENT_MASTER) {
sdsfreesplitres(argv,argc);
......@@ -2292,8 +2295,12 @@ int processMultibulkBuffer(client *c) {
}
c->qb_pos = newline-c->querybuf+2;
if (ll >= PROTO_MBULK_BIG_ARG) {
/* If we are going to read a large object from network
if (!(c->flags & CLIENT_MASTER) && ll >= PROTO_MBULK_BIG_ARG) {
/* When the client is not a master client (because master
* client's querybuf can only be trimmed after data applied
* and sent to replicas).
*
* If we are going to read a large object from network
* try to make it likely that it will start at c->querybuf
* boundary so that we can optimize object creation
* avoiding a large copy of data.
......@@ -2324,10 +2331,11 @@ int processMultibulkBuffer(client *c) {
c->argv = zrealloc(c->argv, sizeof(robj*)*c->argv_len);
}
/* Optimization: if the buffer contains JUST our bulk element
/* Optimization: if a non-master client's buffer contains JUST our bulk element
* instead of creating a new object by *copying* the sds we
* just use the current sds string. */
if (c->qb_pos == 0 &&
if (!(c->flags & CLIENT_MASTER) &&
c->qb_pos == 0 &&
c->bulklen >= PROTO_MBULK_BIG_ARG &&
sdslen(c->querybuf) == (size_t)(c->bulklen+2))
{
......@@ -2388,8 +2396,8 @@ void commandProcessed(client *c) {
if (c->flags & CLIENT_MASTER) {
long long applied = c->reploff - prev_offset;
if (applied) {
replicationFeedStreamFromMasterStream(c->pending_querybuf,applied);
sdsrange(c->pending_querybuf,applied,-1);
replicationFeedStreamFromMasterStream(c->querybuf+c->repl_applied,applied);
c->repl_applied += applied;
}
}
}
......@@ -2432,13 +2440,22 @@ int processCommandAndResetClient(client *c) {
/* This function will execute any fully parsed commands pending on
* the client. Returns C_ERR if the client is no longer valid after executing
* the command, and C_OK for all other cases. */
int processPendingCommandsAndResetClient(client *c) {
int processPendingCommandAndInputBuffer(client *c) {
if (c->flags & CLIENT_PENDING_COMMAND) {
c->flags &= ~CLIENT_PENDING_COMMAND;
if (processCommandAndResetClient(c) == C_ERR) {
return C_ERR;
}
}
/* Now process client if it has more data in it's buffer.
*
* Note: when a master client steps into this function,
* it can always satisfy this condition, because its querbuf
* contains data not applied. */
if (c->querybuf && sdslen(c->querybuf) > 0) {
return processInputBuffer(c);
}
return C_OK;
}
......@@ -2510,8 +2527,26 @@ int processInputBuffer(client *c) {
}
}
/* Trim to pos */
if (c->qb_pos) {
if (c->flags & CLIENT_MASTER) {
/* If the client is a master, trim the querybuf to repl_applied,
* since master client is very special, its querybuf not only
* used to parse command, but also proxy to sub-replicas.
*
* Here are some scenarios we cannot trim to qb_pos:
* 1. we don't receive complete command from master
* 2. master client blocked cause of client pause
* 3. io threads operate read, master client flagged with CLIENT_PENDING_COMMAND
*
* In these scenarios, qb_pos points to the part of the current command
* or the beginning of next command, and the current command is not applied yet,
* so the repl_applied is not equal to qb_pos. */
if (c->repl_applied) {
sdsrange(c->querybuf,c->repl_applied,-1);
c->qb_pos -= c->repl_applied;
c->repl_applied = 0;
}
} else if (c->qb_pos) {
/* Trim to pos */
sdsrange(c->querybuf,c->qb_pos,-1);
c->qb_pos = 0;
}
......@@ -2519,7 +2554,8 @@ int processInputBuffer(client *c) {
/* Update client memory usage after processing the query buffer, this is
* important in case the query buffer is big and wasn't drained during
* the above loop (because of partially sent big commands). */
updateClientMemUsage(c);
if (io_threads_op == IO_THREADS_OP_IDLE)
updateClientMemUsage(c);
return C_OK;
}
......@@ -2546,16 +2582,22 @@ void readQueryFromClient(connection *conn) {
if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1
&& c->bulklen >= PROTO_MBULK_BIG_ARG)
{
ssize_t remaining = (size_t)(c->bulklen+2)-sdslen(c->querybuf);
ssize_t remaining = (size_t)(c->bulklen+2)-(sdslen(c->querybuf)-c->qb_pos);
big_arg = 1;
/* Note that the 'remaining' variable may be zero in some edge case,
* for example once we resume a blocked client after CLIENT PAUSE. */
if (remaining > 0) readlen = remaining;
/* Master client needs expand the readlen when meet BIG_ARG(see #9100),
* but doesn't need align to the next arg, we can read more data. */
if (c->flags & CLIENT_MASTER && readlen < PROTO_IOBUF_LEN)
readlen = PROTO_IOBUF_LEN;
}
qblen = sdslen(c->querybuf);
if (big_arg || sdsalloc(c->querybuf) < PROTO_IOBUF_LEN) {
if (!(c->flags & CLIENT_MASTER) && // master client's querybuf can grow greedy.
(big_arg || sdsalloc(c->querybuf) < PROTO_IOBUF_LEN)) {
/* When reading a BIG_ARG we won't be reading more than that one arg
* into the query buffer, so we don't need to pre-allocate more than we
* need, so using the non-greedy growing. For an initial allocation of
......@@ -2585,12 +2627,6 @@ void readQueryFromClient(connection *conn) {
}
freeClientAsync(c);
goto done;
} else if (c->flags & CLIENT_MASTER) {
/* Append the query buffer to the pending (not applied) buffer
* of the master. We'll use this buffer later in order to have a
* copy of the string applied by the last command executed. */
c->pending_querybuf = sdscatlen(c->pending_querybuf,
c->querybuf+qblen,nread);
}
sdsIncrLen(c->querybuf,nread);
......@@ -2868,8 +2904,6 @@ void clientCommand(client *c) {
" Control the replies sent to the current connection.",
"SETNAME <name>",
" Assign the name <name> to the current connection.",
"GETNAME",
" Get the name of the current connection.",
"UNBLOCK <clientid> [TIMEOUT|ERROR]",
" Unblock the specified blocked client.",
"TRACKING (ON|OFF) [REDIRECT <id>] [BCAST] [PREFIX <prefix> [...]]",
......@@ -3075,6 +3109,10 @@ NULL
if (getLongLongFromObjectOrReply(c,c->argv[2],&id,NULL)
!= C_OK) return;
struct client *target = lookupClientByID(id);
/* Note that we never try to unblock a client blocked on a module command, which
* doesn't have a timeout callback (even in the case of UNBLOCK ERROR).
* The reason is that we assume that if a command doesn't expect to be timedout,
* it also doesn't expect to be unblocked by CLIENT UNBLOCK */
if (target && target->flags & CLIENT_BLOCKED && moduleBlockedClientMayTimeout(target)) {
if (unblock_error)
addReplyError(target,
......@@ -3464,7 +3502,11 @@ static void retainOriginalCommandVector(client *c) {
* original_argv array. */
void redactClientCommandArgument(client *c, int argc) {
retainOriginalCommandVector(c);
decrRefCount(c->argv[argc]);
if (c->original_argv[argc] == shared.redacted) {
/* This argument has already been redacted */
return;
}
decrRefCount(c->original_argv[argc]);
c->original_argv[argc] = shared.redacted;
}
......@@ -4165,8 +4207,8 @@ int handleClientsWithPendingWritesUsingThreads(void) {
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
/* Update the client in the mem usage buckets after we're done processing it in the io-threads */
updateClientMemUsageBucket(c);
/* Update the client in the mem usage after we're done processing it in the io-threads */
updateClientMemUsage(c);
/* Install the write handler if there are pending writes in some
* of the clients. */
......@@ -4274,17 +4316,10 @@ int handleClientsWithPendingReadsUsingThreads(void) {
continue;
}
/* Once io-threads are idle we can update the client in the mem usage buckets */
updateClientMemUsageBucket(c);
if (processPendingCommandsAndResetClient(c) == C_ERR) {
/* If the client is no longer valid, we avoid
* processing the client later. So we just go
* to the next. */
continue;
}
/* Once io-threads are idle we can update the client in the mem usage */
updateClientMemUsage(c);
if (processInputBuffer(c) == C_ERR) {
if (processPendingCommandAndInputBuffer(c) == C_ERR) {
/* If the client is no longer valid, we avoid
* processing the client later. So we just go
* to the next. */
......
......@@ -1532,7 +1532,7 @@ NULL
} else if (!strcasecmp(c->argv[1]->ptr,"stats") && c->argc == 2) {
struct redisMemOverhead *mh = getMemoryOverheadData();
addReplyMapLen(c,26+mh->num_dbs);
addReplyMapLen(c,27+mh->num_dbs);
addReplyBulkCString(c,"peak.allocated");
addReplyLongLong(c,mh->peak_allocated);
......@@ -1552,6 +1552,9 @@ NULL
addReplyBulkCString(c,"clients.normal");
addReplyLongLong(c,mh->clients_normal);
addReplyBulkCString(c,"cluster.links");
addReplyLongLong(c,mh->cluster_links);
addReplyBulkCString(c,"aof.buffer");
addReplyLongLong(c,mh->aof_buffer);
......
......@@ -38,7 +38,7 @@
*/
#include <sys/types.h>
#include <stdint.h>
#include <errno.h>
#include <stdlib.h>
......@@ -62,7 +62,7 @@ static inline void swapfunc (char *, char *, size_t, int);
} while (--i > 0); \
}
#define SWAPINIT(a, es) swaptype = ((char *)a - (char *)0) % sizeof(long) || \
#define SWAPINIT(a, es) swaptype = (uintptr_t)a % sizeof(long) || \
es % sizeof(long) ? 2 : es == sizeof(long)? 0 : 1;
static inline void
......
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