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

Release Redis 7.2 RC3

parents a51eb05b c7b3ce90
......@@ -219,7 +219,7 @@ int serverPubsubShardSubscriptionCount(void) {
/* Return the number of channels + patterns a client is subscribed to. */
int clientSubscriptionsCount(client *c) {
return dictSize(c->pubsub_channels) + listLength(c->pubsub_patterns);
return dictSize(c->pubsub_channels) + dictSize(c->pubsub_patterns);
}
/* Return the number of shard level channels a client is subscribed to. */
......@@ -345,9 +345,8 @@ int pubsubSubscribePattern(client *c, robj *pattern) {
list *clients;
int retval = 0;
if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {
if (dictAdd(c->pubsub_patterns, pattern, NULL) == DICT_OK) {
retval = 1;
listAddNodeTail(c->pubsub_patterns,pattern);
incrRefCount(pattern);
/* Add the client to the pattern -> list of clients hash table */
de = dictFind(server.pubsub_patterns,pattern);
......@@ -374,9 +373,8 @@ int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) {
int retval = 0;
incrRefCount(pattern); /* Protect the object. May be the same we remove */
if ((ln = listSearchKey(c->pubsub_patterns,pattern)) != NULL) {
if (dictDelete(c->pubsub_patterns, pattern) == DICT_OK) {
retval = 1;
listDelNode(c->pubsub_patterns,ln);
/* Remove the client from the pattern -> clients list hash table */
de = dictFind(server.pubsub_patterns,pattern);
serverAssertWithInfo(c,NULL,de != NULL);
......@@ -448,16 +446,20 @@ void pubsubUnsubscribeShardChannels(robj **channels, unsigned int count) {
/* Unsubscribe from all the patterns. Return the number of patterns the
* client was subscribed from. */
int pubsubUnsubscribeAllPatterns(client *c, int notify) {
listNode *ln;
listIter li;
int count = 0;
listRewind(c->pubsub_patterns,&li);
while ((ln = listNext(&li)) != NULL) {
robj *pattern = ln->value;
if (dictSize(c->pubsub_patterns) > 0) {
dictIterator *di = dictGetSafeIterator(c->pubsub_patterns);
dictEntry *de;
count += pubsubUnsubscribePattern(c,pattern,notify);
while ((de = dictNext(di)) != NULL) {
robj *pattern = dictGetKey(de);
count += pubsubUnsubscribePattern(c, pattern, notify);
}
dictReleaseIterator(di);
}
/* We were subscribed to nothing? Still reply to the client. */
if (notify && count == 0) addReplyPubsubPatUnsubscribed(c,NULL);
return count;
}
......@@ -743,7 +745,7 @@ void sunsubscribeCommand(client *c) {
size_t pubsubMemOverhead(client *c) {
/* PubSub patterns */
size_t mem = listLength(c->pubsub_patterns) * sizeof(listNode);
size_t mem = dictMemUsage(c->pubsub_patterns);
/* Global PubSub channels */
mem += dictMemUsage(c->pubsub_channels);
/* Sharded PubSub channels */
......
......@@ -326,7 +326,7 @@ void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags, size_t *lenptr) {
} else if (encode) {
return createStringObjectFromLongLongForValue(val);
} else {
return createObject(OBJ_STRING,sdsfromlonglong(val));
return createStringObjectFromLongLongWithSds(val);
}
}
......@@ -1844,7 +1844,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
if (rdbtype == RDB_TYPE_STRING) {
/* Read string value */
if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
o = tryObjectEncoding(o);
o = tryObjectEncodingEx(o, 0);
} else if (rdbtype == RDB_TYPE_LIST) {
/* Read list value */
if ((len = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL;
......
......@@ -1614,7 +1614,9 @@ usage:
" -a <password> Password for Redis Auth\n"
" --user <username> Used to send ACL style 'AUTH username pass'. Needs -a.\n"
" -u <uri> Server URI.\n"
" -c <clients> Number of parallel connections (default 50)\n"
" -c <clients> Number of parallel connections (default 50).\n"
" Note: If --cluster is used then number of clients has to be\n"
" the same or higher than the number of nodes.\n"
" -n <requests> Total number of requests (default 100000)\n"
" -d <size> Data size of SET/GET value in bytes (default 3)\n"
" --dbnum <db> SELECT the specified db number (default 0)\n"
......
......@@ -1259,6 +1259,7 @@ REDISMODULE_API RedisModuleString * (*RedisModule_CommandFilterArgGet)(RedisModu
REDISMODULE_API int (*RedisModule_CommandFilterArgInsert)(RedisModuleCommandFilterCtx *fctx, int pos, RedisModuleString *arg) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_CommandFilterArgReplace)(RedisModuleCommandFilterCtx *fctx, int pos, RedisModuleString *arg) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_CommandFilterArgDelete)(RedisModuleCommandFilterCtx *fctx, int pos) REDISMODULE_ATTR;
REDISMODULE_API unsigned long long (*RedisModule_CommandFilterGetClientId)(RedisModuleCommandFilterCtx *fctx) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_Fork)(RedisModuleForkDoneHandler cb, void *user_data) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_SendChildHeartbeat)(double progress) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_ExitFromChild)(int retcode) REDISMODULE_ATTR;
......@@ -1619,6 +1620,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(CommandFilterArgInsert);
REDISMODULE_GET_API(CommandFilterArgReplace);
REDISMODULE_GET_API(CommandFilterArgDelete);
REDISMODULE_GET_API(CommandFilterGetClientId);
REDISMODULE_GET_API(Fork);
REDISMODULE_GET_API(SendChildHeartbeat);
REDISMODULE_GET_API(ExitFromChild);
......
......@@ -418,14 +418,14 @@ void feedReplicationBuffer(char *s, size_t len) {
}
if (add_new_block) {
createReplicationBacklogIndex(listLast(server.repl_buffer_blocks));
}
/* Try to trim replication backlog since replication backlog may exceed
* our setting when we add replication stream. Note that it is important to
* try to trim at least one node since in the common case this is where one
* new backlog node is added and one should be removed. See also comments
* in freeMemoryGetNotCountedMemory for details. */
/* It is important to trim after adding replication data to keep the backlog size close to
* repl_backlog_size in the common case. We wait until we add a new block to avoid repeated
* unnecessary trimming attempts when small amounts of data are added. See comments in
* freeMemoryGetNotCountedMemory() for details on replication backlog memory tracking. */
incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
}
}
}
/* Propagate write commands to replication stream.
......
......@@ -292,6 +292,7 @@ void scriptKill(client *c, int is_eval) {
if (mustObeyClient(curr_run_ctx->original_client)) {
addReplyError(c,
"-UNKILLABLE The busy script was sent by a master instance in the context of replication and cannot be killed.");
return;
}
if (curr_run_ctx->flags & SCRIPT_WRITE_DIRTY) {
addReplyError(c,
......
......@@ -38,6 +38,7 @@
#include <limits.h>
#include "sds.h"
#include "sdsalloc.h"
#include "util.h"
const char *SDS_NOINIT = "SDS_NOINIT";
......@@ -526,90 +527,13 @@ sds sdscpy(sds s, const char *t) {
return sdscpylen(s, t, strlen(t));
}
/* Helper for sdscatlonglong() doing the actual number -> string
* conversion. 's' must point to a string with room for at least
* SDS_LLSTR_SIZE bytes.
*
* The function returns the length of the null-terminated string
* representation stored at 's'. */
#define SDS_LLSTR_SIZE 21
int sdsll2str(char *s, long long value) {
char *p, aux;
unsigned long long v;
size_t l;
/* Generate the string representation, this method produces
* a reversed string. */
if (value < 0) {
/* Since v is unsigned, if value==LLONG_MIN, -LLONG_MIN will overflow. */
if (value != LLONG_MIN) {
v = -value;
} else {
v = ((unsigned long long)LLONG_MAX) + 1;
}
} else {
v = value;
}
p = s;
do {
*p++ = '0'+(v%10);
v /= 10;
} while(v);
if (value < 0) *p++ = '-';
/* Compute length and add null term. */
l = p-s;
*p = '\0';
/* Reverse the string. */
p--;
while(s < p) {
aux = *s;
*s = *p;
*p = aux;
s++;
p--;
}
return l;
}
/* Identical sdsll2str(), but for unsigned long long type. */
int sdsull2str(char *s, unsigned long long v) {
char *p, aux;
size_t l;
/* Generate the string representation, this method produces
* a reversed string. */
p = s;
do {
*p++ = '0'+(v%10);
v /= 10;
} while(v);
/* Compute length and add null term. */
l = p-s;
*p = '\0';
/* Reverse the string. */
p--;
while(s < p) {
aux = *s;
*s = *p;
*p = aux;
s++;
p--;
}
return l;
}
/* Create an sds string from a long long value. It is much faster than:
*
* sdscatprintf(sdsempty(),"%lld\n", value);
*/
sds sdsfromlonglong(long long value) {
char buf[SDS_LLSTR_SIZE];
int len = sdsll2str(buf,value);
char buf[LONG_STR_SIZE];
int len = ll2string(buf,sizeof(buf),value);
return sdsnewlen(buf,len);
}
......@@ -745,8 +669,8 @@ sds sdscatfmt(sds s, char const *fmt, ...) {
else
num = va_arg(ap,long long);
{
char buf[SDS_LLSTR_SIZE];
l = sdsll2str(buf,num);
char buf[LONG_STR_SIZE];
l = ll2string(buf,sizeof(buf),num);
if (sdsavail(s) < l) {
s = sdsMakeRoomFor(s,l);
}
......@@ -762,8 +686,8 @@ sds sdscatfmt(sds s, char const *fmt, ...) {
else
unum = va_arg(ap,unsigned long long);
{
char buf[SDS_LLSTR_SIZE];
l = sdsull2str(buf,unum);
char buf[LONG_STR_SIZE];
l = ull2string(buf,sizeof(buf),unum);
if (sdsavail(s) < l) {
s = sdsMakeRoomFor(s,l);
}
......
......@@ -3178,6 +3178,13 @@ void sentinelSendPeriodicCommands(sentinelRedisInstance *ri) {
}
/* =========================== SENTINEL command ============================= */
static void populateDict(dict *options_dict, char **options) {
for (int i=0; options[i]; i++) {
sds option = sdsnew(options[i]);
if (dictAdd(options_dict, option, NULL)==DICT_ERR)
sdsfree(option);
}
}
const char* getLogLevel(void) {
switch (server.verbosity) {
......@@ -3185,56 +3192,120 @@ const char* getLogLevel(void) {
case LL_VERBOSE: return "verbose";
case LL_NOTICE: return "notice";
case LL_WARNING: return "warning";
case LL_NOTHING: return "nothing";
}
return "unknown";
}
/* SENTINEL CONFIG SET <option> <value>*/
/* SENTINEL CONFIG SET option value [option value ...] */
void sentinelConfigSetCommand(client *c) {
robj *o = c->argv[3];
robj *val = c->argv[4];
long long numval;
int drop_conns = 0;
char *option;
robj *val;
char *options[] = {
"announce-ip",
"sentinel-user",
"sentinel-pass",
"resolve-hostnames",
"announce-port",
"announce-hostnames",
"loglevel",
NULL};
static dict *options_dict = NULL;
if (!options_dict) {
options_dict = dictCreate(&stringSetDictType);
populateDict(options_dict, options);
}
dict *set_configs = dictCreate(&stringSetDictType);
/* Validate arguments are valid */
for (int i = 3; i < c->argc; i++) {
option = c->argv[i]->ptr;
/* Validate option is valid */
if (dictFind(options_dict, option) == NULL) {
addReplyErrorFormat(c, "Invalid argument '%s' to SENTINEL CONFIG SET", option);
goto exit;
}
/* Check duplicates */
if (dictFind(set_configs, option) != NULL) {
addReplyErrorFormat(c, "Duplicate argument '%s' to SENTINEL CONFIG SET", option);
goto exit;
}
serverAssert(dictAdd(set_configs, sdsnew(option), NULL) == C_OK);
/* Validate argument */
if (i + 1 == c->argc) {
addReplyErrorFormat(c, "Missing argument '%s' value", option);
goto exit;
}
val = c->argv[++i];
if (!strcasecmp(o->ptr, "resolve-hostnames")) {
if ((numval = yesnotoi(val->ptr)) == -1) goto badfmt;
if (!strcasecmp(option, "resolve-hostnames")) {
if ((yesnotoi(val->ptr)) == -1) goto badfmt;
} else if (!strcasecmp(option, "announce-hostnames")) {
if ((yesnotoi(val->ptr)) == -1) goto badfmt;
} else if (!strcasecmp(option, "announce-port")) {
if (getLongLongFromObject(val, &numval) == C_ERR ||
numval < 0 || numval > 65535) goto badfmt;
} else if (!strcasecmp(option, "loglevel")) {
if (!(!strcasecmp(val->ptr, "debug") || !strcasecmp(val->ptr, "verbose") ||
!strcasecmp(val->ptr, "notice") || !strcasecmp(val->ptr, "warning") ||
!strcasecmp(val->ptr, "nothing"))) goto badfmt;
}
}
/* Apply changes */
for (int i = 3; i < c->argc; i++) {
int moreargs = (c->argc-1) - i;
option = c->argv[i]->ptr;
if (!strcasecmp(option, "loglevel") && moreargs > 0) {
val = c->argv[++i];
if (!strcasecmp(val->ptr, "debug"))
server.verbosity = LL_DEBUG;
else if (!strcasecmp(val->ptr, "verbose"))
server.verbosity = LL_VERBOSE;
else if (!strcasecmp(val->ptr, "notice"))
server.verbosity = LL_NOTICE;
else if (!strcasecmp(val->ptr, "warning"))
server.verbosity = LL_WARNING;
else if (!strcasecmp(val->ptr, "nothing"))
server.verbosity = LL_NOTHING;
} else if (!strcasecmp(option, "resolve-hostnames") && moreargs > 0) {
val = c->argv[++i];
numval = yesnotoi(val->ptr);
sentinel.resolve_hostnames = numval;
} else if (!strcasecmp(o->ptr, "announce-hostnames")) {
if ((numval = yesnotoi(val->ptr)) == -1) goto badfmt;
} else if (!strcasecmp(option, "announce-hostnames") && moreargs > 0) {
val = c->argv[++i];
numval = yesnotoi(val->ptr);
sentinel.announce_hostnames = numval;
} else if (!strcasecmp(o->ptr, "announce-ip")) {
} else if (!strcasecmp(option, "announce-ip") && moreargs > 0) {
val = c->argv[++i];
if (sentinel.announce_ip) sdsfree(sentinel.announce_ip);
sentinel.announce_ip = sdsnew(val->ptr);
} else if (!strcasecmp(o->ptr, "announce-port")) {
if (getLongLongFromObject(val, &numval) == C_ERR ||
numval < 0 || numval > 65535)
goto badfmt;
} else if (!strcasecmp(option, "announce-port") && moreargs > 0) {
val = c->argv[++i];
getLongLongFromObject(val, &numval);
sentinel.announce_port = numval;
} else if (!strcasecmp(o->ptr, "sentinel-user")) {
} else if (!strcasecmp(option, "sentinel-user") && moreargs > 0) {
val = c->argv[++i];
sdsfree(sentinel.sentinel_auth_user);
sentinel.sentinel_auth_user = sdslen(val->ptr) == 0 ?
NULL : sdsdup(val->ptr);
drop_conns = 1;
} else if (!strcasecmp(o->ptr, "sentinel-pass")) {
} else if (!strcasecmp(option, "sentinel-pass") && moreargs > 0) {
val = c->argv[++i];
sdsfree(sentinel.sentinel_auth_pass);
sentinel.sentinel_auth_pass = sdslen(val->ptr) == 0 ?
NULL : sdsdup(val->ptr);
drop_conns = 1;
} else if (!strcasecmp(o->ptr, "loglevel")) {
if (!strcasecmp(val->ptr, "debug"))
server.verbosity = LL_DEBUG;
else if (!strcasecmp(val->ptr, "verbose"))
server.verbosity = LL_VERBOSE;
else if (!strcasecmp(val->ptr, "notice"))
server.verbosity = LL_NOTICE;
else if (!strcasecmp(val->ptr, "warning"))
server.verbosity = LL_WARNING;
else
goto badfmt;
} else {
addReplyErrorFormat(c, "Invalid argument '%s' to SENTINEL CONFIG SET",
(char *) o->ptr);
return;
/* Should never reach here */
serverAssert(0);
}
}
sentinelFlushConfigAndReply(c);
......@@ -3243,61 +3314,72 @@ void sentinelConfigSetCommand(client *c) {
if (drop_conns)
sentinelDropConnections();
exit:
dictRelease(set_configs);
return;
badfmt:
addReplyErrorFormat(c, "Invalid value '%s' to SENTINEL CONFIG SET '%s'",
(char *) val->ptr, (char *) o->ptr);
(char *) val->ptr, option);
dictRelease(set_configs);
}
/* SENTINEL CONFIG GET <option> */
/* SENTINEL CONFIG GET <option> [<option> ...] */
void sentinelConfigGetCommand(client *c) {
robj *o = c->argv[3];
const char *pattern = o->ptr;
char *pattern;
void *replylen = addReplyDeferredLen(c);
int matches = 0;
if (stringmatch(pattern,"resolve-hostnames",1)) {
/* Create a dictionary to store the input configs,to avoid adding duplicate twice */
dict *d = dictCreate(&externalStringType);
for (int i = 3; i < c->argc; i++) {
pattern = c->argv[i]->ptr;
/* If the string doesn't contain glob patterns and available in dictionary, don't look further, just continue. */
if (!strpbrk(pattern, "[*?") && dictFind(d, pattern)) continue;
/* we want to print all the matched patterns and avoid printing duplicates twice */
if (stringmatch(pattern,"resolve-hostnames",1) && !dictFind(d, "resolve-hostnames")) {
addReplyBulkCString(c,"resolve-hostnames");
addReplyBulkCString(c,sentinel.resolve_hostnames ? "yes" : "no");
dictAdd(d, "resolve-hostnames", NULL);
matches++;
}
if (stringmatch(pattern, "announce-hostnames", 1)) {
if (stringmatch(pattern, "announce-hostnames", 1) && !dictFind(d, "announce-hostnames")) {
addReplyBulkCString(c,"announce-hostnames");
addReplyBulkCString(c,sentinel.announce_hostnames ? "yes" : "no");
dictAdd(d, "announce-hostnames", NULL);
matches++;
}
if (stringmatch(pattern, "announce-ip", 1)) {
if (stringmatch(pattern, "announce-ip", 1) && !dictFind(d, "announce-ip")) {
addReplyBulkCString(c,"announce-ip");
addReplyBulkCString(c,sentinel.announce_ip ? sentinel.announce_ip : "");
dictAdd(d, "announce-ip", NULL);
matches++;
}
if (stringmatch(pattern, "announce-port", 1)) {
if (stringmatch(pattern, "announce-port", 1) && !dictFind(d, "announce-port")) {
addReplyBulkCString(c, "announce-port");
addReplyBulkLongLong(c, sentinel.announce_port);
dictAdd(d, "announce-port", NULL);
matches++;
}
if (stringmatch(pattern, "sentinel-user", 1)) {
if (stringmatch(pattern, "sentinel-user", 1) && !dictFind(d, "sentinel-user")) {
addReplyBulkCString(c, "sentinel-user");
addReplyBulkCString(c, sentinel.sentinel_auth_user ? sentinel.sentinel_auth_user : "");
dictAdd(d, "sentinel-user", NULL);
matches++;
}
if (stringmatch(pattern, "sentinel-pass", 1)) {
if (stringmatch(pattern, "sentinel-pass", 1) && !dictFind(d, "sentinel-pass")) {
addReplyBulkCString(c, "sentinel-pass");
addReplyBulkCString(c, sentinel.sentinel_auth_pass ? sentinel.sentinel_auth_pass : "");
dictAdd(d, "sentinel-pass", NULL);
matches++;
}
if (stringmatch(pattern, "loglevel", 1)) {
if (stringmatch(pattern, "loglevel", 1) && !dictFind(d, "loglevel")) {
addReplyBulkCString(c, "loglevel");
addReplyBulkCString(c, getLogLevel());
dictAdd(d, "loglevel", NULL);
matches++;
}
}
dictRelease(d);
setDeferredMapLen(c, replylen, matches);
}
......@@ -3787,9 +3869,9 @@ void sentinelCommand(client *c) {
" Check if the current Sentinel configuration is able to reach the quorum",
" needed to failover a master and the majority needed to authorize the",
" failover.",
"CONFIG SET <param> <value>",
"CONFIG SET param value [param value ...]",
" Set a global Sentinel configuration parameter.",
"CONFIG GET <param>",
"CONFIG GET <param> [param param param ...]",
" Get global Sentinel configuration parameter.",
"DEBUG [<param> <value> ...]",
" Show a list of configurable time parameters and their values (milliseconds).",
......@@ -4042,12 +4124,12 @@ NULL
sentinelSetCommand(c);
} else if (!strcasecmp(c->argv[1]->ptr,"config")) {
if (c->argc < 4) goto numargserr;
if (!strcasecmp(c->argv[2]->ptr,"set") && c->argc == 5)
if (!strcasecmp(c->argv[2]->ptr,"set") && c->argc >= 5)
sentinelConfigSetCommand(c);
else if (!strcasecmp(c->argv[2]->ptr,"get") && c->argc == 4)
else if (!strcasecmp(c->argv[2]->ptr,"get") && c->argc >= 4)
sentinelConfigGetCommand(c);
else
addReplyError(c, "Only SENTINEL CONFIG GET <option> / SET <option> <value> are supported.");
addReplyError(c, "Only SENTINEL CONFIG GET <param> [<param> <param> ...]/ SET <param> <value> [<param> <value> ...] are supported.");
} else if (!strcasecmp(c->argv[1]->ptr,"info-cache")) {
/* SENTINEL INFO-CACHE <name> */
if (c->argc < 2) goto numargserr;
......
......@@ -1324,8 +1324,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
*
* Note that you can change the resolution altering the
* LRU_CLOCK_RESOLUTION define. */
unsigned int lruclock = getLRUClock();
atomicSet(server.lruclock,lruclock);
server.lruclock = getLRUClock();
cronUpdateMemoryStats();
......@@ -1649,9 +1648,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
return;
}
/* Handle precise timeouts of blocked clients. */
handleBlockedClientsTimeout();
/* We should handle pending reads clients ASAP after event loop. */
handleClientsWithPendingReadsUsingThreads();
......@@ -1661,40 +1657,33 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
/* If any connection type(typical TLS) still has pending unread data don't sleep at all. */
aeSetDontWait(server.el, connTypeHasPendingData());
/* Record cron time in beforeSleep, which is the sum of active-expire, active-defrag and all other
* tasks done by cron and beforeSleep, but excluding read, write and AOF, that are counted by other
* sets of metrics. */
monotime cron_start_time_before_aof = getMonotonicUs();
/* Call the Redis Cluster before sleep function. Note that this function
* may change the state of Redis Cluster (from ok to fail or vice versa),
* so it's a good idea to call it before serving the unblocked clients
* later in this function. */
* later in this function, must be done before blockedBeforeSleep. */
if (server.cluster_enabled) clusterBeforeSleep();
/* Handle blocked clients.
* must be done before flushAppendOnlyFile, in case of appendfsync=always,
* since the unblocked clients may write data. */
blockedBeforeSleep();
/* Record cron time in beforeSleep, which is the sum of active-expire, active-defrag and all other
* tasks done by cron and beforeSleep, but excluding read, write and AOF, that are counted by other
* sets of metrics. */
monotime cron_start_time_before_aof = getMonotonicUs();
/* Run a fast expire cycle (the called function will return
* ASAP if a fast cycle is not needed). */
if (server.active_expire_enabled && iAmMaster())
activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
/* Unblock all the clients blocked for synchronous replication
* in WAIT or WAITAOF. */
if (listLength(server.clients_waiting_acks))
processClientsWaitingReplicas();
/* Check if there are clients unblocked by modules that implement
* blocking commands. */
if (moduleCount()) {
moduleFireServerEvent(REDISMODULE_EVENT_EVENTLOOP,
REDISMODULE_SUBEVENT_EVENTLOOP_BEFORE_SLEEP,
NULL);
moduleHandleBlockedClients();
}
/* Try to process pending commands for clients that were just unblocked. */
if (listLength(server.unblocked_clients))
processUnblockedClients();
/* Send all the slaves an ACK request if at least one client blocked
* during the previous event loop iteration. Note that we do this after
* processUnblockedClients(), so if there are multiple pipelined WAITs
......@@ -1718,20 +1707,12 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
* we have to flush them after each command, so when we get here, the list
* must be empty. */
serverAssert(listLength(server.tracking_pending_keys) == 0);
serverAssert(listLength(server.pending_push_messages) == 0);
/* Send the invalidation messages to clients participating to the
* client side caching protocol in broadcasting (BCAST) mode. */
trackingBroadcastInvalidationMessages();
/* Try to process blocked clients every once in while.
*
* Example: A module calls RM_SignalKeyAsReady from within a timer callback
* (So we don't visit processCommand() at all).
*
* must be done before flushAppendOnlyFile, in case of appendfsync=always,
* since the unblocked clients may write data. */
handleClientsBlockedOnKeys();
/* Record time consumption of AOF writing. */
monotime aof_start_time = getMonotonicUs();
/* Record cron time in beforeSleep. This does not include the time consumed by AOF writing and IO writing below. */
......@@ -1985,6 +1966,7 @@ void createSharedObjects(void) {
for (j = 0; j < OBJ_SHARED_INTEGERS; j++) {
shared.integers[j] =
makeObjectShared(createObject(OBJ_STRING,(void*)(long)j));
initObjectLRUOrLFU(shared.integers[j]);
shared.integers[j]->encoding = OBJ_ENCODING_INT;
}
for (j = 0; j < OBJ_SHARED_BULKHDR_LEN; j++) {
......@@ -2089,8 +2071,7 @@ void initServerConfig(void) {
server.latency_tracking_info_percentiles[1] = 99.0; /* p99 */
server.latency_tracking_info_percentiles[2] = 99.9; /* p999 */
unsigned int lruclock = getLRUClock();
atomicSet(server.lruclock,lruclock);
server.lruclock = getLRUClock();
resetServerSaveParams();
appendServerSaveParams(60*60,1); /* save after 1 hour and 1 change */
......@@ -2612,6 +2593,7 @@ void initServer(void) {
server.unblocked_clients = listCreate();
server.ready_keys = listCreate();
server.tracking_pending_keys = listCreate();
server.pending_push_messages = listCreate();
server.clients_waiting_acks = listCreate();
server.get_ack_from_slaves = 0;
server.paused_actions = 0;
......@@ -3715,9 +3697,11 @@ void call(client *c, int flags) {
* various pre-execution checks. it returns the appropriate error to the client.
* If there's a transaction is flags it as dirty, and if the command is EXEC,
* it aborts the transaction.
* The duration is reset, since we reject the command, and it did not record.
* Note: 'reply' is expected to end with \r\n */
void rejectCommand(client *c, robj *reply) {
flagTransaction(c);
c->duration = 0;
if (c->cmd) c->cmd->rejected_calls++;
if (c->cmd && c->cmd->proc == execCommand) {
execCommandAbort(c, reply->ptr);
......@@ -3729,6 +3713,7 @@ void rejectCommand(client *c, robj *reply) {
void rejectCommandSds(client *c, sds s) {
flagTransaction(c);
c->duration = 0;
if (c->cmd) c->cmd->rejected_calls++;
if (c->cmd && c->cmd->proc == execCommand) {
execCommandAbort(c, s);
......@@ -3756,7 +3741,14 @@ void afterCommand(client *c) {
/* Should be done before trackingHandlePendingKeyInvalidations so that we
* reply to client before invalidating cache (makes more sense) */
postExecutionUnitOperations();
/* Flush pending tracking invalidations. */
trackingHandlePendingKeyInvalidations();
/* Flush other pending push messages. only when we are not in nested call.
* So the messages are not interleaved with transaction response. */
if (!server.execution_nesting)
listJoin(c->reply, server.pending_push_messages);
}
/* Check if c->cmd exists, fills `err` with details in case it doesn't.
......@@ -5496,8 +5488,6 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) {
call_uname = 0;
}
unsigned int lruclock;
atomicGet(server.lruclock,lruclock);
info = sdscatfmt(info,
"# Server\r\n"
"redis_version:%s\r\n"
......@@ -5548,7 +5538,7 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) {
(int64_t)(uptime/(3600*24)),
server.hz,
server.config_hz,
lruclock,
server.lruclock,
server.executable ? server.executable : "",
server.configfile ? server.configfile : "",
server.io_threads_active);
......
......@@ -507,6 +507,7 @@ typedef enum {
#define LL_VERBOSE 1
#define LL_NOTICE 2
#define LL_WARNING 3
#define LL_NOTHING 4
#define LL_RAW (1<<10) /* Modifier to log without timestamp */
/* Supervision options */
......@@ -591,9 +592,9 @@ typedef enum {
#define CMD_CALL_PROPAGATE_AOF (1<<0)
#define CMD_CALL_PROPAGATE_REPL (1<<1)
#define CMD_CALL_REPROCESSING (1<<2)
#define CMD_CALL_FROM_MODULE (1<<3) /* From RM_Call */
#define CMD_CALL_PROPAGATE (CMD_CALL_PROPAGATE_AOF|CMD_CALL_PROPAGATE_REPL)
#define CMD_CALL_FULL (CMD_CALL_PROPAGATE)
#define CMD_CALL_FROM_MODULE (1<<2) /* From RM_Call */
/* Command propagation flags, see propagateNow() function */
#define PROPAGATE_NONE 0
......@@ -709,6 +710,7 @@ typedef enum {
* encoding version. */
#define OBJ_MODULE 5 /* Module object. */
#define OBJ_STREAM 6 /* Stream object. */
#define OBJ_TYPE_MAX 7 /* Maximum number of object types */
/* Extract encver / signature from a module type ID. */
#define REDISMODULE_TYPE_ENCVER_BITS 10
......@@ -1214,7 +1216,7 @@ typedef struct client {
long long woff; /* Last write global replication offset. */
list *watched_keys; /* Keys WATCHED for MULTI/EXEC CAS */
dict *pubsub_channels; /* channels a client is interested in (SUBSCRIBE) */
list *pubsub_patterns; /* patterns a client is interested in (SUBSCRIBE) */
dict *pubsub_patterns; /* patterns a client is interested in (PSUBSCRIBE) */
dict *pubsubshard_channels; /* shard level channels a client is interested in (SSUBSCRIBE) */
sds peerid; /* Cached peer ID. */
sds sockname; /* Cached connection target address. */
......@@ -1548,7 +1550,7 @@ struct redisServer {
dict *orig_commands; /* Command table before command renaming. */
aeEventLoop *el;
rax *errors; /* Errors table */
redisAtomic unsigned int lruclock; /* Clock for LRU eviction */
unsigned int lruclock; /* Clock for LRU eviction */
volatile sig_atomic_t shutdown_asap; /* Shutdown ordered by signal handler. */
mstime_t shutdown_mstime; /* Timestamp to limit graceful shutdown. */
int last_sig_received; /* Indicates the last SIGNAL received, if any (e.g., SIGINT or SIGTERM). */
......@@ -1927,6 +1929,7 @@ struct redisServer {
unsigned int tracking_clients; /* # of clients with tracking enabled.*/
size_t tracking_table_max_keys; /* Max number of keys in tracking table. */
list *tracking_pending_keys; /* tracking invalidation keys pending to flush */
list *pending_push_messages; /* pending publish or other push messages to flush */
/* Sort parameters - qsort_r() is only available under BSD so we
* have to take this state global, in order to pass it to sortCompare() */
int sort_desc;
......@@ -1978,6 +1981,7 @@ struct redisServer {
if the master is in failure state. */
char *cluster_announce_ip; /* IP address to announce on cluster bus. */
char *cluster_announce_hostname; /* hostname to announce on cluster bus. */
char *cluster_announce_human_nodename; /* Human readable node name assigned to a node. */
int cluster_preferred_endpoint_type; /* Use the announced hostname when available. */
int cluster_announce_port; /* base port to announce on cluster bus. */
int cluster_announce_tls_port; /* TLS port to announce on cluster bus. */
......@@ -2468,6 +2472,8 @@ void moduleLoadFromQueue(void);
int moduleGetCommandKeysViaAPI(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int moduleGetCommandChannelsViaAPI(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
moduleType *moduleTypeLookupModuleByID(uint64_t id);
moduleType *moduleTypeLookupModuleByName(const char *name);
moduleType *moduleTypeLookupModuleByNameIgnoreCase(const char *name);
void moduleTypeNameByID(char *name, uint64_t moduleid);
const char *moduleTypeModuleName(moduleType *mt);
const char *moduleNameFromCommand(struct redisCommand *cmd);
......@@ -2727,6 +2733,7 @@ void freeZsetObject(robj *o);
void freeHashObject(robj *o);
void dismissObject(robj *o, size_t dump_size);
robj *createObject(int type, void *ptr);
void initObjectLRUOrLFU(robj *o);
robj *createStringObject(const char *ptr, size_t len);
robj *createRawStringObject(const char *ptr, size_t len);
robj *createEmbeddedStringObject(const char *ptr, size_t len);
......@@ -2736,10 +2743,12 @@ robj *dupStringObject(const robj *o);
int isSdsRepresentableAsLongLong(sds s, long long *llval);
int isObjectRepresentableAsLongLong(robj *o, long long *llongval);
robj *tryObjectEncoding(robj *o);
robj *tryObjectEncodingEx(robj *o, int try_trim);
robj *getDecodedObject(robj *o);
size_t stringObjectLen(robj *o);
robj *createStringObjectFromLongLong(long long value);
robj *createStringObjectFromLongLongForValue(long long value);
robj *createStringObjectFromLongLongWithSds(long long value);
robj *createStringObjectFromLongDouble(long double value, int humanfriendly);
robj *createQuicklistObject(void);
robj *createListListpackObject(void);
......@@ -3092,7 +3101,6 @@ void setTypeReleaseIterator(setTypeIterator *si);
int setTypeNext(setTypeIterator *si, char **str, size_t *len, int64_t *llele);
sds setTypeNextObject(setTypeIterator *si);
int setTypeRandomElement(robj *setobj, char **str, size_t *len, int64_t *llele);
unsigned long setTypeRandomElements(robj *set, unsigned long count, robj *aux_set);
unsigned long setTypeSize(const robj *subject);
void setTypeConvert(robj *subject, int enc);
int setTypeConvertAndExpand(robj *setobj, int enc, unsigned long cap, int panic);
......@@ -3246,6 +3254,7 @@ void dbReplaceValue(redisDb *db, robj *key, robj *val);
#define SETKEY_NO_SIGNAL 2
#define SETKEY_ALREADY_EXIST 4
#define SETKEY_DOESNT_EXIST 8
#define SETKEY_ADD_OR_UPDATE 16 /* Key most likely doesn't exists */
void setKey(client *c, redisDb *db, robj *key, robj *val, int flags);
robj *dbRandomKey(redisDb *db);
int dbGenericDelete(redisDb *db, robj *key, int async, int flags);
......@@ -3377,7 +3386,7 @@ void signalDeletedKeyAsReady(redisDb *db, robj *key, int type);
void updateStatsOnUnblock(client *c, long blocked_us, long reply_us, int had_errors);
void scanDatabaseForDeletedKeys(redisDb *emptied, redisDb *replaced_with);
void totalNumberOfBlockingKeys(unsigned long *blocking_keys, unsigned long *bloking_keys_on_nokey);
void blockedBeforeSleep(void);
/* timeout.c -- Blocked clients timeout and connections timeout. */
void addClientToTimeoutTable(client *c);
......
......@@ -78,6 +78,7 @@ static connection *connCreateSocket(void) {
connection *conn = zcalloc(sizeof(connection));
conn->type = &CT_Socket;
conn->fd = -1;
conn->iovcnt = IOV_MAX;
return conn;
}
......
......@@ -45,7 +45,7 @@ void hashTypeTryConversion(robj *o, robj **argv, int start, int end) {
/* We guess that most of the values in the input are unique, so
* if there are enough arguments we create a pre-sized hash, which
* might overallocate memory if their are duplicates. */
* might over allocate memory if there are duplicates. */
size_t new_fields = (end - start + 1) / 2;
if (new_fields > server.hash_max_listpack_entries) {
hashTypeConvert(o, OBJ_ENCODING_HT);
......@@ -1014,6 +1014,26 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
return;
}
/* CASE 2.5 listpack only. Sampling unique elements, in non-random order.
* Listpack encoded hashes are meant to be relatively small, so
* HRANDFIELD_SUB_STRATEGY_MUL isn't necessary and we rather not make
* copies of the entries. Instead, we emit them directly to the output
* buffer.
*
* And it is inefficient to repeatedly pick one random element from a
* listpack in CASE 4. So we use this instead. */
if (hash->encoding == OBJ_ENCODING_LISTPACK) {
listpackEntry *keys, *vals = NULL;
keys = zmalloc(sizeof(listpackEntry)*count);
if (withvalues)
vals = zmalloc(sizeof(listpackEntry)*count);
serverAssert(lpRandomPairsUnique(hash->ptr, count, keys, vals) == count);
hrandfieldReplyWithListpack(c, count, keys, vals);
zfree(keys);
zfree(vals);
return;
}
/* CASE 3:
* The number of elements inside the hash is not greater than
* HRANDFIELD_SUB_STRATEGY_MUL times the number of requested elements.
......@@ -1024,6 +1044,7 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
* a bit less than the number of elements in the hash, the natural approach
* used into CASE 4 is highly inefficient. */
if (count*HRANDFIELD_SUB_STRATEGY_MUL > size) {
/* Hashtable encoding (generic implementation) */
dict *d = dictCreate(&sdsReplyDictType);
dictExpand(d, size);
hashTypeIterator *hi = hashTypeInitIterator(hash);
......@@ -1077,20 +1098,6 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
* to the temporary hash, trying to eventually get enough unique elements
* to reach the specified count. */
else {
if (hash->encoding == OBJ_ENCODING_LISTPACK) {
/* it is inefficient to repeatedly pick one random element from a
* listpack. so we use this instead: */
listpackEntry *keys, *vals = NULL;
keys = zmalloc(sizeof(listpackEntry)*count);
if (withvalues)
vals = zmalloc(sizeof(listpackEntry)*count);
serverAssert(lpRandomPairsUnique(hash->ptr, count, keys, vals) == count);
hrandfieldReplyWithListpack(c, count, keys, vals);
zfree(keys);
zfree(vals);
return;
}
/* Hashtable encoding (generic implementation) */
unsigned long added = 0;
listpackEntry key, value;
......
......@@ -240,7 +240,7 @@ listTypeIterator *listTypeInitIterator(robj *subject, long index,
li->direction = direction;
li->iter = NULL;
/* LIST_HEAD means start at TAIL and move *towards* head.
* LIST_TAIL means start at HEAD and move *towards tail. */
* LIST_TAIL means start at HEAD and move *towards* tail. */
if (li->encoding == OBJ_ENCODING_QUICKLIST) {
int iter_direction = direction == LIST_HEAD ? AL_START_TAIL : AL_START_HEAD;
li->iter = quicklistGetIteratorAtIdx(li->subject->ptr,
......
......@@ -38,19 +38,19 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
robj *dstkey, int op);
/* Factory method to return a set that *can* hold "value". When the object has
* an integer-encodable value, an intset will be returned. Otherwise a regular
* hash table.
* an integer-encodable value, an intset will be returned. Otherwise a listpack
* or a regular hash table.
*
* The size hint indicates approximately how many items will be added which is
* used to determine the initial representation. */
robj *setTypeCreate(sds value, size_t size_hint) {
if (isSdsRepresentableAsLongLong(value,NULL) == C_OK && size_hint < server.set_max_intset_entries)
if (isSdsRepresentableAsLongLong(value,NULL) == C_OK && size_hint <= server.set_max_intset_entries)
return createIntsetObject();
if (size_hint < server.set_max_listpack_entries)
if (size_hint <= server.set_max_listpack_entries)
return createSetListpackObject();
/* We may oversize the set by using the hint if the hint is not accurate,
* but we will assume this is accpetable to maximize performance. */
* but we will assume this is acceptable to maximize performance. */
robj *o = createSetObject();
dictExpand(o->ptr, size_hint);
return o;
......@@ -59,8 +59,8 @@ robj *setTypeCreate(sds value, size_t size_hint) {
/* Check if the existing set should be converted to another encoding based off the
* the size hint. */
void setTypeMaybeConvert(robj *set, size_t size_hint) {
if ((set->encoding == OBJ_ENCODING_LISTPACK && size_hint >= server.set_max_listpack_entries)
|| (set->encoding == OBJ_ENCODING_INTSET && size_hint >= server.set_max_intset_entries))
if ((set->encoding == OBJ_ENCODING_LISTPACK && size_hint > server.set_max_listpack_entries)
|| (set->encoding == OBJ_ENCODING_INTSET && size_hint > server.set_max_intset_entries))
{
setTypeConvertAndExpand(set, OBJ_ENCODING_HT, size_hint, 1);
}
......@@ -798,8 +798,9 @@ void spopWithCountCommand(client *c) {
/* todo: Move the spop notification to be executed after the command logic. */
/* Propagate this command as a DEL operation */
rewriteClientCommandVector(c,2,shared.del,c->argv[1]);
/* Propagate this command as a DEL or UNLINK operation */
robj *aux = server.lazyfree_lazy_server_del ? shared.unlink : shared.del;
rewriteClientCommandVector(c, 2, aux, c->argv[1]);
signalModifiedKey(c,c->db,c->argv[1]);
return;
}
......@@ -807,13 +808,14 @@ void spopWithCountCommand(client *c) {
/* Case 2 and 3 require to replicate SPOP as a set of SREM commands.
* Prepare our replication argument vector. Also send the array length
* which is common to both the code paths. */
robj *propargv[3];
unsigned long batchsize = count > 1024 ? 1024 : count;
robj **propargv = zmalloc(sizeof(robj *) * (2 + batchsize));
propargv[0] = shared.srem;
propargv[1] = c->argv[1];
unsigned long propindex = 2;
addReplySetLen(c,count);
/* Common iteration vars. */
robj *objele;
char *str;
size_t len;
int64_t llele;
......@@ -841,16 +843,19 @@ void spopWithCountCommand(client *c) {
if (str) {
addReplyBulkCBuffer(c, str, len);
objele = createStringObject(str, len);
propargv[propindex++] = createStringObject(str, len);
} else {
addReplyBulkLongLong(c, llele);
objele = createStringObjectFromLongLong(llele);
propargv[propindex++] = createStringObjectFromLongLong(llele);
}
/* Replicate/AOF this command as an SREM operation */
propargv[2] = objele;
alsoPropagate(c->db->id,propargv,3,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(objele);
if (propindex == 2 + batchsize) {
alsoPropagate(c->db->id, propargv, propindex, PROPAGATE_AOF | PROPAGATE_REPL);
for (unsigned long j = 2; j < propindex; j++) {
decrRefCount(propargv[j]);
}
propindex = 2;
}
/* Store pointer for later deletion and move to next. */
ps[i] = p;
......@@ -861,14 +866,18 @@ void spopWithCountCommand(client *c) {
zfree(ps);
set->ptr = lp;
} else if (remaining*SPOP_MOVE_STRATEGY_MUL > count) {
while(count--) {
objele = setTypePopRandom(set);
addReplyBulk(c, objele);
for (unsigned long i = 0; i < count; i++) {
propargv[propindex] = setTypePopRandom(set);
addReplyBulk(c, propargv[propindex]);
propindex++;
/* Replicate/AOF this command as an SREM operation */
propargv[2] = objele;
alsoPropagate(c->db->id,propargv,3,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(objele);
if (propindex == 2 + batchsize) {
alsoPropagate(c->db->id, propargv, propindex, PROPAGATE_AOF | PROPAGATE_REPL);
for (unsigned long j = 2; j < propindex; j++) {
decrRefCount(propargv[j]);
}
propindex = 2;
}
}
} else {
/* CASE 3: The number of elements to return is very big, approaching
......@@ -918,16 +927,19 @@ void spopWithCountCommand(client *c) {
while (setTypeNext(si, &str, &len, &llele) != -1) {
if (str == NULL) {
addReplyBulkLongLong(c,llele);
objele = createStringObjectFromLongLong(llele);
propargv[propindex++] = createStringObjectFromLongLong(llele);
} else {
addReplyBulkCBuffer(c, str, len);
objele = createStringObject(str, len);
propargv[propindex++] = createStringObject(str, len);
}
/* Replicate/AOF this command as an SREM operation */
propargv[2] = objele;
alsoPropagate(c->db->id,propargv,3,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(objele);
if (propindex == 2 + batchsize) {
alsoPropagate(c->db->id, propargv, propindex, PROPAGATE_AOF | PROPAGATE_REPL);
for (unsigned long i = 2; i < propindex; i++) {
decrRefCount(propargv[i]);
}
propindex = 2;
}
}
setTypeReleaseIterator(si);
......@@ -935,6 +947,16 @@ void spopWithCountCommand(client *c) {
dbReplaceValue(c->db,c->argv[1],newset);
}
/* Replicate/AOF the remaining elements as an SREM operation */
if (propindex != 2) {
alsoPropagate(c->db->id, propargv, propindex, PROPAGATE_AOF | PROPAGATE_REPL);
for (unsigned long i = 2; i < propindex; i++) {
decrRefCount(propargv[i]);
}
propindex = 2;
}
zfree(propargv);
/* Don't propagate the command itself even if we incremented the
* dirty counter. We don't want to propagate an SPOP command since
* we propagated the command as a set of SREMs operations using
......@@ -1093,7 +1115,10 @@ void srandmemberWithCountCommand(client *c) {
* Listpack encoded sets are meant to be relatively small, so
* SRANDMEMBER_SUB_STRATEGY_MUL isn't necessary and we rather not make
* copies of the entries. Instead, we emit them directly to the output
* buffer. */
* buffer.
*
* And it is inefficient to repeatedly pick one random element from a
* listpack in CASE 4. So we use this instead. */
if (set->encoding == OBJ_ENCODING_LISTPACK) {
unsigned char *lp = set->ptr;
unsigned char *p = lpFirst(lp);
......@@ -1316,7 +1341,7 @@ void sinterGenericCommand(client *c, robj **setkeys,
} else if (sets[0]->encoding == OBJ_ENCODING_LISTPACK) {
/* To avoid many reallocs, we estimate that the result is a listpack
* of approximately the same size as the first set. Then we shrink
* it or possibly convert it to intset it in the end. */
* it or possibly convert it to intset in the end. */
unsigned char *lp = lpNew(lpBytes(sets[0]->ptr));
dstset = createObject(OBJ_SET, lp);
dstset->encoding = OBJ_ENCODING_LISTPACK;
......@@ -1516,8 +1541,8 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
}
}
/* We need a temp set object to store our union. If the dstkey
* is not NULL (that is, we are inside an SUNIONSTORE operation) then
/* We need a temp set object to store our union/diff. If the dstkey
* is not NULL (that is, we are inside an SUNIONSTORE/SDIFFSTORE operation) then
* this set object will be the resulting object to set into the target key*/
dstset = createIntsetObject();
......
......@@ -461,7 +461,7 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
}
/* Avoid overflow when trying to add an element to the stream (listpack
* can only host up to 32bit length sttrings, and also a total listpack size
* can only host up to 32bit length strings, and also a total listpack size
* can't be bigger than 32bit length. */
size_t totelelen = 0;
for (int64_t i = 0; i < numfields*2; i++) {
......@@ -2208,9 +2208,10 @@ void xreadCommand(client *c) {
streams_arg = i+1;
streams_count = (c->argc-streams_arg);
if ((streams_count % 2) != 0) {
char symbol = xreadgroup ? '>' : '$';
addReplyErrorFormat(c,"Unbalanced '%s' list of streams: "
"for each stream key an ID or '>' must be "
"specified.", c->cmd->fullname);
"for each stream key an ID or '%c' must be "
"specified.", c->cmd->fullname,symbol);
return;
}
streams_count /= 2; /* We have two arguments for each stream. */
......
......@@ -116,10 +116,12 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
if (expire) {
setExpire(c,c->db,key,milliseconds);
/* Propagate as SET Key Value PXAT millisecond-timestamp if there is
* EX/PX/EXAT/PXAT flag. */
* EX/PX/EXAT flag. */
if (!(flags & OBJ_PXAT)) {
robj *milliseconds_obj = createStringObjectFromLongLong(milliseconds);
rewriteClientCommandVector(c, 5, shared.set, key, val, shared.pxat, milliseconds_obj);
decrRefCount(milliseconds_obj);
}
notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);
}
......@@ -388,8 +390,7 @@ void getexCommand(client *c) {
if (((flags & OBJ_PXAT) || (flags & OBJ_EXAT)) && checkAlreadyExpired(milliseconds)) {
/* When PXAT/EXAT absolute timestamp is specified, there can be a chance that timestamp
* has already elapsed so delete the key in that case. */
int deleted = server.lazyfree_lazy_expire ? dbAsyncDelete(c->db, c->argv[1]) :
dbSyncDelete(c->db, c->argv[1]);
int deleted = dbGenericDelete(c->db, c->argv[1], server.lazyfree_lazy_expire, DB_FLAG_KEY_EXPIRED);
serverAssert(deleted);
robj *aux = server.lazyfree_lazy_expire ? shared.unlink : shared.del;
rewriteClientCommandVector(c,2,aux,c->argv[1]);
......@@ -576,10 +577,14 @@ void msetGenericCommand(client *c, int nx) {
}
}
int setkey_flags = nx ? SETKEY_DOESNT_EXIST : 0;
for (j = 1; j < c->argc; j += 2) {
c->argv[j+1] = tryObjectEncoding(c->argv[j+1]);
setKey(c, c->db, c->argv[j], c->argv[j + 1], 0);
setKey(c, c->db, c->argv[j], c->argv[j + 1], setkey_flags);
notifyKeyspaceEvent(NOTIFY_STRING,"set",c->argv[j],c->db->id);
/* In MSETNX, It could be that we're overriding the same key, we can't be sure it doesn't exist. */
if (nx)
setkey_flags = SETKEY_ADD_OR_UPDATE;
}
server.dirty += (c->argc-1)/2;
addReply(c, nx ? shared.cone : shared.ok);
......
......@@ -66,6 +66,7 @@
int zslLexValueGteMin(sds value, zlexrangespec *spec);
int zslLexValueLteMax(sds value, zlexrangespec *spec);
void zsetConvertAndExpand(robj *zobj, int encoding, unsigned long cap);
/* Create a skiplist node with the specified number of levels.
* The SDS string 'ele' is referenced by the node after the call. */
......@@ -1165,7 +1166,45 @@ unsigned long zsetLength(const robj *zobj) {
return length;
}
/* Factory method to return a zset.
*
* The size hint indicates approximately how many items will be added,
* and the value len hint indicates the approximate individual size of the added elements,
* they are used to determine the initial representation.
*
* If the hints are not known, and underestimation or 0 is suitable. */
robj *zsetTypeCreate(size_t size_hint, size_t val_len_hint) {
if (size_hint <= server.zset_max_listpack_entries &&
val_len_hint <= server.zset_max_listpack_value)
{
return createZsetListpackObject();
}
robj *zobj = createZsetObject();
zset *zs = zobj->ptr;
dictExpand(zs->dict, size_hint);
return zobj;
}
/* Check if the existing zset should be converted to another encoding based off the
* the size hint. */
void zsetTypeMaybeConvert(robj *zobj, size_t size_hint) {
if (zobj->encoding == OBJ_ENCODING_LISTPACK &&
size_hint > server.zset_max_listpack_entries)
{
zsetConvertAndExpand(zobj, OBJ_ENCODING_SKIPLIST, size_hint);
}
}
/* Convert the zset to specified encoding. The zset dict (when converting
* to a skiplist) is presized to hold the number of elements in the original
* zset. */
void zsetConvert(robj *zobj, int encoding) {
zsetConvertAndExpand(zobj, encoding, zsetLength(zobj));
}
/* Converts a zset to the specified encoding, pre-sizing it for 'cap' elements. */
void zsetConvertAndExpand(robj *zobj, int encoding, unsigned long cap) {
zset *zs;
zskiplistNode *node, *next;
sds ele;
......@@ -1186,6 +1225,9 @@ void zsetConvert(robj *zobj, int encoding) {
zs->dict = dictCreate(&zsetDictType);
zs->zsl = zslCreate();
/* Presize the dict to avoid rehashing */
dictExpand(zs->dict, cap);
eptr = lpSeek(zl,0);
if (eptr != NULL) {
sptr = lpNext(zl,eptr);
......@@ -1375,7 +1417,7 @@ int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, dou
sdslen(ele) > server.zset_max_listpack_value ||
!lpSafeToAdd(zobj->ptr, sdslen(ele)))
{
zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
zsetConvertAndExpand(zobj, OBJ_ENCODING_SKIPLIST, zsetLength(zobj) + 1);
} else {
zobj->ptr = zzlInsert(zobj->ptr,ele,score);
if (newscore) *newscore = score;
......@@ -1749,14 +1791,10 @@ void zaddGenericCommand(client *c, int flags) {
if (checkType(c,zobj,OBJ_ZSET)) goto cleanup;
if (zobj == NULL) {
if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */
if (server.zset_max_listpack_entries == 0 ||
server.zset_max_listpack_value < sdslen(c->argv[scoreidx+1]->ptr))
{
zobj = createZsetObject();
} else {
zobj = createZsetListpackObject();
}
zobj = zsetTypeCreate(elements, sdslen(c->argv[scoreidx+1]->ptr));
dbAdd(c->db,key,zobj);
} else {
zsetTypeMaybeConvert(zobj, elements);
}
for (j = 0; j < elements; j++) {
......@@ -2537,8 +2575,8 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in
zsetopval zval;
sds tmp;
size_t maxelelen = 0, totelelen = 0;
robj *dstobj;
zset *dstzset;
robj *dstobj = NULL;
zset *dstzset = NULL;
zskiplistNode *znode;
int withscores = 0;
unsigned long cardinality = 0;
......@@ -2648,8 +2686,14 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in
qsort(src,setnum,sizeof(zsetopsrc),zuiCompareByCardinality);
}
/* We need a temp zset object to store our union/inter/diff. If the dstkey
* is not NULL (that is, we are inside an ZUNIONSTORE/ZINTERSTORE/ZDIFFSTORE operation) then
* this zset object will be the resulting object to zset into the target key.
* In SINTERCARD case, we don't need the temp obj, so we can avoid creating it. */
if (!cardinality_only) {
dstobj = createZsetObject();
dstzset = dstobj->ptr;
}
memset(&zval, 0, sizeof(zval));
if (op == SET_OP_INTER) {
......@@ -2788,6 +2832,7 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in
server.dirty++;
}
}
decrRefCount(dstobj);
} else if (cardinality_only) {
addReplyLongLong(c, cardinality);
} else {
......@@ -2808,8 +2853,9 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in
if (withscores) addReplyDouble(c,zn->score);
zn = zn->level[0].forward;
}
}
server.lazyfree_lazy_server_del ? freeObjAsync(NULL, dstobj, -1) :
decrRefCount(dstobj);
}
zfree(src);
}
......@@ -2955,10 +3001,7 @@ static void zrangeResultFinalizeClient(zrange_result_handler *handler,
/* Result handler methods for storing the ZRANGESTORE to a zset. */
static void zrangeResultBeginStore(zrange_result_handler *handler, long length)
{
if (length > (long)server.zset_max_listpack_entries)
handler->dstobj = createZsetObject();
else
handler->dstobj = createZsetListpackObject();
handler->dstobj = zsetTypeCreate(length, 0);
}
static void zrangeResultEmitCBufferForStore(zrange_result_handler *handler,
......@@ -4208,6 +4251,27 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
return;
}
/* CASE 2.5 listpack only. Sampling unique elements, in non-random order.
* Listpack encoded zsets are meant to be relatively small, so
* ZRANDMEMBER_SUB_STRATEGY_MUL isn't necessary and we rather not make
* copies of the entries. Instead, we emit them directly to the output
* buffer.
*
* And it is inefficient to repeatedly pick one random element from a
* listpack in CASE 4. So we use this instead. */
if (zsetobj->encoding == OBJ_ENCODING_LISTPACK) {
listpackEntry *keys, *vals = NULL;
keys = zmalloc(sizeof(listpackEntry)*count);
if (withscores)
vals = zmalloc(sizeof(listpackEntry)*count);
serverAssert(lpRandomPairsUnique(zsetobj->ptr, count, keys, vals) == count);
zrandmemberReplyWithListpack(c, count, keys, vals);
zfree(keys);
zfree(vals);
zuiClearIterator(&src);
return;
}
/* CASE 3:
* The number of elements inside the zset is not greater than
* ZRANDMEMBER_SUB_STRATEGY_MUL times the number of requested elements.
......@@ -4218,6 +4282,7 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
* a bit less than the number of elements in the set, the natural approach
* used into CASE 4 is highly inefficient. */
if (count*ZRANDMEMBER_SUB_STRATEGY_MUL > size) {
/* Hashtable encoding (generic implementation) */
dict *d = dictCreate(&sdsReplyDictType);
dictExpand(d, size);
/* Add all the elements into the temporary dictionary. */
......@@ -4261,21 +4326,6 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
* to the temporary set, trying to eventually get enough unique elements
* to reach the specified count. */
else {
if (zsetobj->encoding == OBJ_ENCODING_LISTPACK) {
/* it is inefficient to repeatedly pick one random element from a
* listpack. so we use this instead: */
listpackEntry *keys, *vals = NULL;
keys = zmalloc(sizeof(listpackEntry)*count);
if (withscores)
vals = zmalloc(sizeof(listpackEntry)*count);
serverAssert(lpRandomPairsUnique(zsetobj->ptr, count, keys, vals) == count);
zrandmemberReplyWithListpack(c, count, keys, vals);
zfree(keys);
zfree(vals);
zuiClearIterator(&src);
return;
}
/* Hashtable encoding (generic implementation) */
unsigned long added = 0;
dict *d = dictCreate(&hashDictType);
......
......@@ -462,6 +462,7 @@ static connection *createTLSConnection(int client_side) {
tls_connection *conn = zcalloc(sizeof(tls_connection));
conn->c.type = &CT_TLS;
conn->c.fd = -1;
conn->c.iovcnt = IOV_MAX;
conn->ssl = SSL_new(ctx);
return (connection *) conn;
}
......
......@@ -78,6 +78,7 @@ static connection *connCreateUnix(void) {
connection *conn = zcalloc(sizeof(connection));
conn->type = &CT_Unix;
conn->fd = -1;
conn->iovcnt = IOV_MAX;
return 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