Commit d7971f96 authored by Oran Agra's avatar Oran Agra
Browse files

Merge remote-tracking branch 'origin/unstable' into 7.0

parents d2b5a579 acfb4f7a
...@@ -482,10 +482,12 @@ void debugCommand(client *c) { ...@@ -482,10 +482,12 @@ void debugCommand(client *c) {
" Show low level client eviction pools info (maxmemory-clients).", " Show low level client eviction pools info (maxmemory-clients).",
"PAUSE-CRON <0|1>", "PAUSE-CRON <0|1>",
" Stop periodic cron job processing.", " 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.", " 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 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", " 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 NULL
}; };
addReplyHelp(c, help); addReplyHelp(c, help);
...@@ -793,6 +795,10 @@ NULL ...@@ -793,6 +795,10 @@ NULL
* also have a normal reply type after the attribute. */ * also have a normal reply type after the attribute. */
addReplyBulkCString(c,"Some real reply following the attribute"); addReplyBulkCString(c,"Some real reply following the attribute");
} else if (!strcasecmp(name,"push")) { } else if (!strcasecmp(name,"push")) {
if (c->resp < 3) {
addReplyError(c,"RESP2 is not supported by this command");
return;
}
addReplyPushLen(c,2); addReplyPushLen(c,2);
addReplyBulkCString(c,"server-cpu-usage"); addReplyBulkCString(c,"server-cpu-usage");
addReplyLongLong(c,42); addReplyLongLong(c,42);
...@@ -962,14 +968,21 @@ NULL ...@@ -962,14 +968,21 @@ NULL
{ {
server.pause_cron = atoi(c->argv[2]->ptr); server.pause_cron = atoi(c->argv[2]->ptr);
addReply(c,shared.ok); addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"replybuffer-peak-reset-time") && c->argc == 3 ) { } else if (!strcasecmp(c->argv[1]->ptr,"replybuffer") && c->argc == 4 ) {
if (!strcasecmp(c->argv[2]->ptr, "never")) { if(!strcasecmp(c->argv[2]->ptr, "peak-reset-time")) {
server.reply_buffer_peak_reset_time = -1; if (!strcasecmp(c->argv[3]->ptr, "never")) {
} else if(!strcasecmp(c->argv[2]->ptr, "reset")) { server.reply_buffer_peak_reset_time = -1;
server.reply_buffer_peak_reset_time = REPLY_BUFFER_DEFAULT_PEAK_RESET_TIME; } 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 { } else {
if (getLongFromObjectOrReply(c, c->argv[2], &server.reply_buffer_peak_reset_time, NULL) != C_OK) addReplySubcommandSyntaxError(c);
return; return;
} }
addReply(c, shared.ok); addReply(c, shared.ok);
} else { } else {
...@@ -1943,7 +1956,7 @@ void sigsegvHandler(int sig, siginfo_t *info, void *secret) { ...@@ -1943,7 +1956,7 @@ void sigsegvHandler(int sig, siginfo_t *info, void *secret) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Accessing address: %p", (void*)info->si_addr); "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); 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) { ...@@ -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. * 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 * 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 * 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 * NOTE: this is very ugly code, but it let's us avoid the complication of
* doing a scan on another dict. */ * doing a scan on another dict. */
dictEntry* replaceSatelliteDictKeyPtrAndOrDefragDictEntry(dict *d, sds oldkey, sds newkey, uint64_t hash, long *defragged) { dictEntry* replaceSatelliteDictKeyPtrAndOrDefragDictEntry(dict *d, sds oldkey, sds newkey, uint64_t hash, long *defragged) {
...@@ -1196,7 +1196,7 @@ void activeDefragCycle(void) { ...@@ -1196,7 +1196,7 @@ void activeDefragCycle(void) {
cursor = dictScan(db->dict, cursor, defragScanCallback, defragDictBucketCallback, db); cursor = dictScan(db->dict, cursor, defragScanCallback, defragDictBucketCallback, db);
/* Once in 16 scan iterations, 512 pointer reallocations. or 64 keys /* 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. * check if we reached the time limit.
* But regardless, don't start a new db in this loop, this is because after * 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 */ * the last db we call defragOtherGlobals, which must be done in one cycle */
......
...@@ -508,6 +508,7 @@ void evalGenericCommand(client *c, int evalsha) { ...@@ -508,6 +508,7 @@ void evalGenericCommand(client *c, int evalsha) {
scriptRunCtx rctx; scriptRunCtx rctx;
if (scriptPrepareForRun(&rctx, lctx.lua_client, c, lua_cur_script, l->flags, ro) != C_OK) { 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; return;
} }
rctx.flags |= SCRIPT_EVAL_MODE; /* mark the current run as EVAL (as opposed to FCALL) so we'll rctx.flags |= SCRIPT_EVAL_MODE; /* mark the current run as EVAL (as opposed to FCALL) so we'll
......
...@@ -57,6 +57,12 @@ struct functionsLibCtx { ...@@ -57,6 +57,12 @@ struct functionsLibCtx {
dict *engines_stats; /* Per engine statistics */ dict *engines_stats; /* Per engine statistics */
}; };
typedef struct functionsLibMataData {
sds engine;
sds name;
sds code;
} functionsLibMataData;
dictType engineDictType = { dictType engineDictType = {
dictSdsCaseHash, /* hash function */ dictSdsCaseHash, /* hash function */
dictSdsDup, /* key dup */ dictSdsDup, /* key dup */
...@@ -124,7 +130,6 @@ static size_t functionMallocSize(functionInfo *fi) { ...@@ -124,7 +130,6 @@ static size_t functionMallocSize(functionInfo *fi) {
static size_t libraryMallocSize(functionLibInfo *li) { static size_t libraryMallocSize(functionLibInfo *li) {
return zmalloc_size(li) + sdsZmallocSize(li->name) return zmalloc_size(li) + sdsZmallocSize(li->name)
+ (li->desc ? sdsZmallocSize(li->desc) : 0)
+ sdsZmallocSize(li->code); + sdsZmallocSize(li->code);
} }
...@@ -157,7 +162,6 @@ static void engineLibraryFree(functionLibInfo* li) { ...@@ -157,7 +162,6 @@ static void engineLibraryFree(functionLibInfo* li) {
dictRelease(li->functions); dictRelease(li->functions);
sdsfree(li->name); sdsfree(li->name);
sdsfree(li->code); sdsfree(li->code);
if (li->desc) sdsfree(li->desc);
zfree(li); zfree(li);
} }
...@@ -265,14 +269,13 @@ int functionLibCreateFunction(sds name, void *function, functionLibInfo *li, sds ...@@ -265,14 +269,13 @@ int functionLibCreateFunction(sds name, void *function, functionLibInfo *li, sds
return C_OK; 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)); functionLibInfo *li = zmalloc(sizeof(*li));
*li = (functionLibInfo) { *li = (functionLibInfo) {
.name = sdsdup(name), .name = sdsdup(name),
.functions = dictCreate(&libraryFunctionDictType), .functions = dictCreate(&libraryFunctionDictType),
.ei = ei, .ei = ei,
.code = sdsdup(code), .code = sdsdup(code),
.desc = desc ? sdsdup(desc) : NULL,
}; };
return li; return li;
} }
...@@ -540,17 +543,11 @@ void functionListCommand(client *c) { ...@@ -540,17 +543,11 @@ void functionListCommand(client *c) {
} }
} }
++reply_len; ++reply_len;
addReplyMapLen(c, with_code? 5 : 4); addReplyMapLen(c, with_code? 4 : 3);
addReplyBulkCString(c, "library_name"); addReplyBulkCString(c, "library_name");
addReplyBulkCBuffer(c, li->name, sdslen(li->name)); addReplyBulkCBuffer(c, li->name, sdslen(li->name));
addReplyBulkCString(c, "engine"); addReplyBulkCString(c, "engine");
addReplyBulkCBuffer(c, li->ei->name, sdslen(li->ei->name)); 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"); addReplyBulkCString(c, "functions");
addReplyArrayLen(c, dictSize(li->functions)); addReplyArrayLen(c, dictSize(li->functions));
...@@ -745,11 +742,11 @@ void functionRestoreCommand(client *c) { ...@@ -745,11 +742,11 @@ void functionRestoreCommand(client *c) {
err = sdsnew("can not read data type"); err = sdsnew("can not read data type");
goto load_error; 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"); err = sdsnew("given type is not a function");
goto load_error; 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) { if (!err) {
err = sdsnew("failed loading the given functions payload"); err = sdsnew("failed loading the given functions payload");
} }
...@@ -868,36 +865,111 @@ static int functionsVerifyName(sds name) { ...@@ -868,36 +865,111 @@ static int functionsVerifyName(sds name) {
return C_OK; return C_OK;
} }
/* Compile and save the given library, return C_OK on success and C_ERR on failure. int functionExtractLibMetaData(sds payload, functionsLibMataData *md, sds *err) {
* In case on failure the err out param is set with relevant error message */ sds name = NULL;
int functionsCreateWithLibraryCtx(sds lib_name,sds engine_name, sds desc, sds code, sds desc = NULL;
int replace, sds* err, functionsLibCtx *lib_ctx) { 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; dictIterator *iter = NULL;
dictEntry *entry = 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"); *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) { if (!ei) {
*err = sdsnew("Engine not found"); *err = sdscatfmt(sdsempty(), "Engine '%S' not found", md.engine);
return C_ERR; goto error;
} }
engine *engine = ei->engine; 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) { if (old_li && !replace) {
*err = sdsnew("Library already exists"); *err = sdscatfmt(sdsempty(), "Library '%S' already exists", md.name);
return C_ERR; goto error;
} }
if (old_li) { if (old_li) {
libraryUnlink(lib_ctx, old_li); libraryUnlink(lib_ctx, old_li);
} }
functionLibInfo *new_li = engineLibraryCreate(lib_name, ei, desc, code); new_li = engineLibraryCreate(md.name, ei, code);
if (engine->create(engine->engine_ctx, new_li, code, err) != C_OK) { if (engine->create(engine->engine_ctx, new_li, md.code, err) != C_OK) {
goto error; goto error;
} }
...@@ -925,48 +997,34 @@ int functionsCreateWithLibraryCtx(sds lib_name,sds engine_name, sds desc, sds co ...@@ -925,48 +997,34 @@ int functionsCreateWithLibraryCtx(sds lib_name,sds engine_name, sds desc, sds co
engineLibraryFree(old_li); engineLibraryFree(old_li);
} }
return C_OK; sds loaded_lib_name = md.name;
md.name = NULL;
functionFreeLibMetaData(&md);
return loaded_lib_name;
error: error:
if (iter) dictReleaseIterator(iter); if (iter) dictReleaseIterator(iter);
engineLibraryFree(new_li); if (new_li) engineLibraryFree(new_li);
if (old_li) { if (old_li) libraryLink(lib_ctx, old_li);
libraryLink(lib_ctx, old_li); functionFreeLibMetaData(&md);
} return NULL;
return C_ERR;
} }
/* /*
* FUNCTION LOAD <ENGINE NAME> <LIBRARY NAME> * FUNCTION LOAD [REPLACE] <LIBRARY CODE>
* [REPLACE] [DESC <LIBRARY DESCRIPTION>] <LIBRARY CODE>
*
* ENGINE NAME - name of the engine to use the run the library
* LIBRARY NAME - name of the library
* REPLACE - optional, replace existing library * REPLACE - optional, replace existing library
* DESCRIPTION - optional, library description
* LIBRARY CODE - library code to pass to the engine * LIBRARY CODE - library code to pass to the engine
*/ */
void functionLoadCommand(client *c) { void functionLoadCommand(client *c) {
robj *engine_name = c->argv[2];
robj *library_name = c->argv[3];
int replace = 0; int replace = 0;
int argc_pos = 4; int argc_pos = 2;
sds desc = NULL;
while (argc_pos < c->argc - 1) { while (argc_pos < c->argc - 1) {
robj *next_arg = c->argv[argc_pos++]; robj *next_arg = c->argv[argc_pos++];
if (!strcasecmp(next_arg->ptr, "replace")) { if (!strcasecmp(next_arg->ptr, "replace")) {
replace = 1; replace = 1;
continue; 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); addReplyErrorFormat(c, "Unknown option given: %s", (char*)next_arg->ptr);
return; return;
} }
...@@ -978,8 +1036,8 @@ void functionLoadCommand(client *c) { ...@@ -978,8 +1036,8 @@ void functionLoadCommand(client *c) {
robj *code = c->argv[argc_pos]; robj *code = c->argv[argc_pos];
sds err = NULL; sds err = NULL;
if (functionsCreateWithLibraryCtx(library_name->ptr, engine_name->ptr, sds library_name = NULL;
desc, code->ptr, replace, &err, curr_functions_lib_ctx) != C_OK) if (!(library_name = functionsCreateWithLibraryCtx(code->ptr, replace, &err, curr_functions_lib_ctx)))
{ {
addReplyErrorSds(c, err); addReplyErrorSds(c, err);
return; return;
...@@ -987,7 +1045,7 @@ void functionLoadCommand(client *c) { ...@@ -987,7 +1045,7 @@ void functionLoadCommand(client *c) {
/* Indicate that the command changed the data so it will be replicated and /* Indicate that the command changed the data so it will be replicated and
* counted as a data change (for persistence configuration) */ * counted as a data change (for persistence configuration) */
server.dirty++; server.dirty++;
addReply(c, shared.ok); addReplyBulkSds(c, library_name);
} }
/* Return memory usage of all the engines combine */ /* Return memory usage of all the engines combine */
......
...@@ -106,12 +106,10 @@ struct functionLibInfo { ...@@ -106,12 +106,10 @@ struct functionLibInfo {
dict *functions; /* Functions dictionary */ dict *functions; /* Functions dictionary */
engineInfo *ei; /* Pointer to the function engine */ engineInfo *ei; /* Pointer to the function engine */
sds code; /* Library code */ sds code; /* Library code */
sds desc; /* Library description */
}; };
int functionsRegisterEngine(const char *engine_name, engine *engine_ctx); int functionsRegisterEngine(const char *engine_name, engine *engine_ctx);
int functionsCreateWithLibraryCtx(sds lib_name, sds engine_name, sds desc, sds code, sds functionsCreateWithLibraryCtx(sds code, int replace, sds* err, functionsLibCtx *lib_ctx);
int replace, sds* err, functionsLibCtx *lib_ctx);
unsigned long functionsMemory(); unsigned long functionsMemory();
unsigned long functionsMemoryOverhead(); unsigned long functionsMemoryOverhead();
unsigned long functionsNum(); unsigned long functionsNum();
......
...@@ -46,7 +46,7 @@ ...@@ -46,7 +46,7 @@
/* Interleave lower bits of x and y, so the bits of x /* Interleave lower bits of x and y, so the bits of x
* are in the even positions and bits from y in the odd; * 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 * From: https://graphics.stanford.edu/~seander/bithacks.html#InterleaveBMN
*/ */
static inline uint64_t interleave64(uint32_t xlo, uint32_t ylo) { static inline uint64_t interleave64(uint32_t xlo, uint32_t ylo) {
......
...@@ -399,6 +399,11 @@ struct commandHelp { ...@@ -399,6 +399,11 @@ struct commandHelp {
"Bind a hash slot to a specific node", "Bind a hash slot to a specific node",
12, 12,
"3.0.0" }, "3.0.0" },
{ "CLUSTER SHARDS",
"",
"Get array of cluster slots to node mappings",
12,
"7.0.0" },
{ "CLUSTER SLAVES", { "CLUSTER SLAVES",
"node-id", "node-id",
"List replica nodes of the specified master node", "List replica nodes of the specified master node",
...@@ -431,7 +436,7 @@ struct commandHelp { ...@@ -431,7 +436,7 @@ struct commandHelp {
"2.8.13" }, "2.8.13" },
{ "COMMAND GETKEYSANDFLAGS", { "COMMAND GETKEYSANDFLAGS",
"", "",
"Extract keys given a full Redis command", "Extract keys and access flags given a full Redis command",
9, 9,
"7.0.0" }, "7.0.0" },
{ "COMMAND HELP", { "COMMAND HELP",
...@@ -630,7 +635,7 @@ struct commandHelp { ...@@ -630,7 +635,7 @@ struct commandHelp {
10, 10,
"7.0.0" }, "7.0.0" },
{ "FUNCTION LOAD", { "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)", "Create a function with the given arguments (name, code, description)",
10, 10,
"7.0.0" }, "7.0.0" },
...@@ -1019,6 +1024,11 @@ struct commandHelp { ...@@ -1019,6 +1024,11 @@ struct commandHelp {
"Load a module", "Load a module",
9, 9,
"4.0.0" }, "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", { "MODULE UNLOAD",
"name", "name",
"Unload a module", "Unload a module",
......
...@@ -958,7 +958,7 @@ unsigned char *lpPrependInteger(unsigned char *lp, long long lval) { ...@@ -958,7 +958,7 @@ unsigned char *lpPrependInteger(unsigned char *lp, long long lval) {
return lpInsertInteger(lp, lval, p, LP_BEFORE, NULL); 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 * listpack. It is implemented in terms of lpInsert(), so the return value is
* the same as lpInsert(). */ * the same as lpInsert(). */
unsigned char *lpAppend(unsigned char *lp, unsigned char *ele, uint32_t size) { 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) { ...@@ -108,7 +108,7 @@ void nolocks_localtime(struct tm *tmp, time_t t, time_t tz, int dst) {
int main(void) { int main(void) {
/* Obtain timezone and daylight info. */ /* Obtain timezone and daylight info. */
tzset(); /* Now 'timezome' global is populated. */ tzset(); /* Now 'timezone' global is populated. */
time_t t = time(NULL); time_t t = time(NULL);
struct tm *aux = localtime(&t); struct tm *aux = localtime(&t);
int daylight_active = aux->tm_isdst; int daylight_active = aux->tm_isdst;
......
This diff is collapsed.
...@@ -31,7 +31,6 @@ ...@@ -31,7 +31,6 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h" #include "../redismodule.h"
#include <pthread.h> #include <pthread.h>
#include <unistd.h> #include <unistd.h>
......
...@@ -31,7 +31,6 @@ ...@@ -31,7 +31,6 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h" #include "../redismodule.h"
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
......
...@@ -30,7 +30,6 @@ ...@@ -30,7 +30,6 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h" #include "../redismodule.h"
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
......
...@@ -33,7 +33,6 @@ ...@@ -33,7 +33,6 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h" #include "../redismodule.h"
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
......
...@@ -30,7 +30,6 @@ ...@@ -30,7 +30,6 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h" #include "../redismodule.h"
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
......
...@@ -30,7 +30,6 @@ ...@@ -30,7 +30,6 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#define REDISMODULE_EXPERIMENTAL_API
#include "../redismodule.h" #include "../redismodule.h"
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
......
...@@ -147,7 +147,6 @@ client *createClient(connection *conn) { ...@@ -147,7 +147,6 @@ client *createClient(connection *conn) {
c->ref_block_pos = 0; c->ref_block_pos = 0;
c->qb_pos = 0; c->qb_pos = 0;
c->querybuf = sdsempty(); c->querybuf = sdsempty();
c->pending_querybuf = sdsempty();
c->querybuf_peak = 0; c->querybuf_peak = 0;
c->reqtype = 0; c->reqtype = 0;
c->argc = 0; c->argc = 0;
...@@ -167,6 +166,7 @@ client *createClient(connection *conn) { ...@@ -167,6 +166,7 @@ client *createClient(connection *conn) {
c->repl_start_cmd_stream_on_ack = 0; c->repl_start_cmd_stream_on_ack = 0;
c->reploff = 0; c->reploff = 0;
c->read_reploff = 0; c->read_reploff = 0;
c->repl_applied = 0;
c->repl_ack_off = 0; c->repl_ack_off = 0;
c->repl_ack_time = 0; c->repl_ack_time = 0;
c->repl_last_partial_write = 0; c->repl_last_partial_write = 0;
...@@ -201,7 +201,7 @@ client *createClient(connection *conn) { ...@@ -201,7 +201,7 @@ client *createClient(connection *conn) {
c->pending_read_list_node = NULL; c->pending_read_list_node = NULL;
c->client_tracking_redirection = 0; c->client_tracking_redirection = 0;
c->client_tracking_prefixes = NULL; 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->last_memory_type = CLIENT_TYPE_NORMAL;
c->auth_callback = NULL; c->auth_callback = NULL;
c->auth_callback_privdata = NULL; c->auth_callback_privdata = NULL;
...@@ -1568,7 +1568,6 @@ void freeClient(client *c) { ...@@ -1568,7 +1568,6 @@ void freeClient(client *c) {
/* Free the query buffer */ /* Free the query buffer */
sdsfree(c->querybuf); sdsfree(c->querybuf);
sdsfree(c->pending_querybuf);
c->querybuf = NULL; c->querybuf = NULL;
/* Deallocate structures used to block on blocking ops. */ /* Deallocate structures used to block on blocking ops. */
...@@ -1940,7 +1939,7 @@ int writeToClient(client *c, int handler_installed) { ...@@ -1940,7 +1939,7 @@ int writeToClient(client *c, int handler_installed) {
if (!clientHasPendingReplies(c)) { if (!clientHasPendingReplies(c)) {
c->sentlen = 0; c->sentlen = 0;
/* Note that writeToClient() is called in a threaded way, but /* 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 * is always called with handler_installed set to 0 from threads
* so we are fine. */ * so we are fine. */
if (handler_installed) { if (handler_installed) {
...@@ -1954,7 +1953,11 @@ int writeToClient(client *c, int handler_installed) { ...@@ -1954,7 +1953,11 @@ int writeToClient(client *c, int handler_installed) {
return C_ERR; 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; return C_OK;
} }
...@@ -2128,7 +2131,7 @@ int processInlineBuffer(client *c) { ...@@ -2128,7 +2131,7 @@ int processInlineBuffer(client *c) {
* we got some desynchronization in the protocol, for example * we got some desynchronization in the protocol, for example
* because of a PSYNC gone bad. * 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. */ * to keep the connection active. */
if (querylen != 0 && c->flags & CLIENT_MASTER) { if (querylen != 0 && c->flags & CLIENT_MASTER) {
sdsfreesplitres(argv,argc); sdsfreesplitres(argv,argc);
...@@ -2292,8 +2295,12 @@ int processMultibulkBuffer(client *c) { ...@@ -2292,8 +2295,12 @@ int processMultibulkBuffer(client *c) {
} }
c->qb_pos = newline-c->querybuf+2; c->qb_pos = newline-c->querybuf+2;
if (ll >= PROTO_MBULK_BIG_ARG) { if (!(c->flags & CLIENT_MASTER) && ll >= PROTO_MBULK_BIG_ARG) {
/* If we are going to read a large object from network /* 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 * try to make it likely that it will start at c->querybuf
* boundary so that we can optimize object creation * boundary so that we can optimize object creation
* avoiding a large copy of data. * avoiding a large copy of data.
...@@ -2324,10 +2331,11 @@ int processMultibulkBuffer(client *c) { ...@@ -2324,10 +2331,11 @@ int processMultibulkBuffer(client *c) {
c->argv = zrealloc(c->argv, sizeof(robj*)*c->argv_len); 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 * instead of creating a new object by *copying* the sds we
* just use the current sds string. */ * 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 && c->bulklen >= PROTO_MBULK_BIG_ARG &&
sdslen(c->querybuf) == (size_t)(c->bulklen+2)) sdslen(c->querybuf) == (size_t)(c->bulklen+2))
{ {
...@@ -2388,8 +2396,8 @@ void commandProcessed(client *c) { ...@@ -2388,8 +2396,8 @@ void commandProcessed(client *c) {
if (c->flags & CLIENT_MASTER) { if (c->flags & CLIENT_MASTER) {
long long applied = c->reploff - prev_offset; long long applied = c->reploff - prev_offset;
if (applied) { if (applied) {
replicationFeedStreamFromMasterStream(c->pending_querybuf,applied); replicationFeedStreamFromMasterStream(c->querybuf+c->repl_applied,applied);
sdsrange(c->pending_querybuf,applied,-1); c->repl_applied += applied;
} }
} }
} }
...@@ -2432,13 +2440,22 @@ int processCommandAndResetClient(client *c) { ...@@ -2432,13 +2440,22 @@ int processCommandAndResetClient(client *c) {
/* This function will execute any fully parsed commands pending on /* 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 client. Returns C_ERR if the client is no longer valid after executing
* the command, and C_OK for all other cases. */ * the command, and C_OK for all other cases. */
int processPendingCommandsAndResetClient(client *c) { int processPendingCommandAndInputBuffer(client *c) {
if (c->flags & CLIENT_PENDING_COMMAND) { if (c->flags & CLIENT_PENDING_COMMAND) {
c->flags &= ~CLIENT_PENDING_COMMAND; c->flags &= ~CLIENT_PENDING_COMMAND;
if (processCommandAndResetClient(c) == C_ERR) { if (processCommandAndResetClient(c) == C_ERR) {
return 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; return C_OK;
} }
...@@ -2510,8 +2527,26 @@ int processInputBuffer(client *c) { ...@@ -2510,8 +2527,26 @@ int processInputBuffer(client *c) {
} }
} }
/* Trim to pos */ if (c->flags & CLIENT_MASTER) {
if (c->qb_pos) { /* 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); sdsrange(c->querybuf,c->qb_pos,-1);
c->qb_pos = 0; c->qb_pos = 0;
} }
...@@ -2519,7 +2554,8 @@ int processInputBuffer(client *c) { ...@@ -2519,7 +2554,8 @@ int processInputBuffer(client *c) {
/* Update client memory usage after processing the query buffer, this is /* Update client memory usage after processing the query buffer, this is
* important in case the query buffer is big and wasn't drained during * important in case the query buffer is big and wasn't drained during
* the above loop (because of partially sent big commands). */ * the above loop (because of partially sent big commands). */
updateClientMemUsage(c); if (io_threads_op == IO_THREADS_OP_IDLE)
updateClientMemUsage(c);
return C_OK; return C_OK;
} }
...@@ -2546,16 +2582,22 @@ void readQueryFromClient(connection *conn) { ...@@ -2546,16 +2582,22 @@ void readQueryFromClient(connection *conn) {
if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1 if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1
&& c->bulklen >= PROTO_MBULK_BIG_ARG) && 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; big_arg = 1;
/* Note that the 'remaining' variable may be zero in some edge case, /* Note that the 'remaining' variable may be zero in some edge case,
* for example once we resume a blocked client after CLIENT PAUSE. */ * for example once we resume a blocked client after CLIENT PAUSE. */
if (remaining > 0) readlen = remaining; 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); 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 /* 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 * 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 * need, so using the non-greedy growing. For an initial allocation of
...@@ -2585,12 +2627,6 @@ void readQueryFromClient(connection *conn) { ...@@ -2585,12 +2627,6 @@ void readQueryFromClient(connection *conn) {
} }
freeClientAsync(c); freeClientAsync(c);
goto done; 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); sdsIncrLen(c->querybuf,nread);
...@@ -2868,8 +2904,6 @@ void clientCommand(client *c) { ...@@ -2868,8 +2904,6 @@ void clientCommand(client *c) {
" Control the replies sent to the current connection.", " Control the replies sent to the current connection.",
"SETNAME <name>", "SETNAME <name>",
" Assign the name <name> to the current connection.", " Assign the name <name> to the current connection.",
"GETNAME",
" Get the name of the current connection.",
"UNBLOCK <clientid> [TIMEOUT|ERROR]", "UNBLOCK <clientid> [TIMEOUT|ERROR]",
" Unblock the specified blocked client.", " Unblock the specified blocked client.",
"TRACKING (ON|OFF) [REDIRECT <id>] [BCAST] [PREFIX <prefix> [...]]", "TRACKING (ON|OFF) [REDIRECT <id>] [BCAST] [PREFIX <prefix> [...]]",
...@@ -3075,6 +3109,10 @@ NULL ...@@ -3075,6 +3109,10 @@ NULL
if (getLongLongFromObjectOrReply(c,c->argv[2],&id,NULL) if (getLongLongFromObjectOrReply(c,c->argv[2],&id,NULL)
!= C_OK) return; != C_OK) return;
struct client *target = lookupClientByID(id); 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 (target && target->flags & CLIENT_BLOCKED && moduleBlockedClientMayTimeout(target)) {
if (unblock_error) if (unblock_error)
addReplyError(target, addReplyError(target,
...@@ -3464,7 +3502,11 @@ static void retainOriginalCommandVector(client *c) { ...@@ -3464,7 +3502,11 @@ static void retainOriginalCommandVector(client *c) {
* original_argv array. */ * original_argv array. */
void redactClientCommandArgument(client *c, int argc) { void redactClientCommandArgument(client *c, int argc) {
retainOriginalCommandVector(c); 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; c->original_argv[argc] = shared.redacted;
} }
...@@ -4165,8 +4207,8 @@ int handleClientsWithPendingWritesUsingThreads(void) { ...@@ -4165,8 +4207,8 @@ int handleClientsWithPendingWritesUsingThreads(void) {
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
client *c = listNodeValue(ln); client *c = listNodeValue(ln);
/* Update the client in the mem usage buckets after we're done processing it in the io-threads */ /* Update the client in the mem usage after we're done processing it in the io-threads */
updateClientMemUsageBucket(c); updateClientMemUsage(c);
/* Install the write handler if there are pending writes in some /* Install the write handler if there are pending writes in some
* of the clients. */ * of the clients. */
...@@ -4274,17 +4316,10 @@ int handleClientsWithPendingReadsUsingThreads(void) { ...@@ -4274,17 +4316,10 @@ int handleClientsWithPendingReadsUsingThreads(void) {
continue; continue;
} }
/* Once io-threads are idle we can update the client in the mem usage buckets */ /* Once io-threads are idle we can update the client in the mem usage */
updateClientMemUsageBucket(c); updateClientMemUsage(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;
}
if (processInputBuffer(c) == C_ERR) { if (processPendingCommandAndInputBuffer(c) == C_ERR) {
/* If the client is no longer valid, we avoid /* If the client is no longer valid, we avoid
* processing the client later. So we just go * processing the client later. So we just go
* to the next. */ * to the next. */
......
...@@ -1532,7 +1532,7 @@ NULL ...@@ -1532,7 +1532,7 @@ NULL
} else if (!strcasecmp(c->argv[1]->ptr,"stats") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"stats") && c->argc == 2) {
struct redisMemOverhead *mh = getMemoryOverheadData(); struct redisMemOverhead *mh = getMemoryOverheadData();
addReplyMapLen(c,26+mh->num_dbs); addReplyMapLen(c,27+mh->num_dbs);
addReplyBulkCString(c,"peak.allocated"); addReplyBulkCString(c,"peak.allocated");
addReplyLongLong(c,mh->peak_allocated); addReplyLongLong(c,mh->peak_allocated);
...@@ -1552,6 +1552,9 @@ NULL ...@@ -1552,6 +1552,9 @@ NULL
addReplyBulkCString(c,"clients.normal"); addReplyBulkCString(c,"clients.normal");
addReplyLongLong(c,mh->clients_normal); addReplyLongLong(c,mh->clients_normal);
addReplyBulkCString(c,"cluster.links");
addReplyLongLong(c,mh->cluster_links);
addReplyBulkCString(c,"aof.buffer"); addReplyBulkCString(c,"aof.buffer");
addReplyLongLong(c,mh->aof_buffer); addReplyLongLong(c,mh->aof_buffer);
......
...@@ -38,7 +38,7 @@ ...@@ -38,7 +38,7 @@
*/ */
#include <sys/types.h> #include <sys/types.h>
#include <stdint.h>
#include <errno.h> #include <errno.h>
#include <stdlib.h> #include <stdlib.h>
...@@ -62,7 +62,7 @@ static inline void swapfunc (char *, char *, size_t, int); ...@@ -62,7 +62,7 @@ static inline void swapfunc (char *, char *, size_t, int);
} while (--i > 0); \ } 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; es % sizeof(long) ? 2 : es == sizeof(long)? 0 : 1;
static inline void static inline void
......
...@@ -905,9 +905,9 @@ int raxInsert(rax *rax, unsigned char *s, size_t len, void *data, void **old) { ...@@ -905,9 +905,9 @@ int raxInsert(rax *rax, unsigned char *s, size_t len, void *data, void **old) {
return raxGenericInsert(rax,s,len,data,old,1); return raxGenericInsert(rax,s,len,data,old,1);
} }
/* Non overwriting insert function: this if an element with the same key /* Non overwriting insert function: if an element with the same key
* exists, the value is not updated and the function returns 0. * exists, the value is not updated and the function returns 0.
* This is a just a wrapper for raxGenericInsert(). */ * This is just a wrapper for raxGenericInsert(). */
int raxTryInsert(rax *rax, unsigned char *s, size_t len, void *data, void **old) { int raxTryInsert(rax *rax, unsigned char *s, size_t len, void *data, void **old) {
return raxGenericInsert(rax,s,len,data,old,0); return raxGenericInsert(rax,s,len,data,old,0);
} }
......
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