Unverified Commit 70469b76 authored by Salvatore Sanfilippo's avatar Salvatore Sanfilippo Committed by GitHub
Browse files

Merge branch 'unstable' into expose_zmalloc_capabilities

parents e45239e3 c697edf4
...@@ -813,11 +813,11 @@ replica-priority 100 ...@@ -813,11 +813,11 @@ replica-priority 100
# MAXMEMORY POLICY: how Redis will select what to remove when maxmemory # MAXMEMORY POLICY: how Redis will select what to remove when maxmemory
# is reached. You can select among five behaviors: # is reached. You can select among five behaviors:
# #
# volatile-lru -> Evict using approximated LRU among the keys with an expire set. # volatile-lru -> Evict using approximated LRU, only keys with an expire set.
# allkeys-lru -> Evict any key using approximated LRU. # allkeys-lru -> Evict any key using approximated LRU.
# volatile-lfu -> Evict using approximated LFU among the keys with an expire set. # volatile-lfu -> Evict using approximated LFU, only keys with an expire set.
# allkeys-lfu -> Evict any key using approximated LFU. # allkeys-lfu -> Evict any key using approximated LFU.
# volatile-random -> Remove a random key among the ones with an expire set. # volatile-random -> Remove a random key having an expire set.
# allkeys-random -> Remove a random key, any key. # allkeys-random -> Remove a random key, any key.
# volatile-ttl -> Remove the key with the nearest expire time (minor TTL) # volatile-ttl -> Remove the key with the nearest expire time (minor TTL)
# noeviction -> Don't evict anything, just return an error on write operations. # noeviction -> Don't evict anything, just return an error on write operations.
...@@ -872,6 +872,23 @@ replica-priority 100 ...@@ -872,6 +872,23 @@ replica-priority 100
# #
# replica-ignore-maxmemory yes # replica-ignore-maxmemory yes
# Redis reclaims expired keys in two ways: upon access when those keys are
# found to be expired, and also in background, in what is called the
# "active expire key". The key space is slowly and interactively scanned
# looking for expired keys to reclaim, so that it is possible to free memory
# of keys that are expired and will never be accessed again in a short time.
#
# The default effort of the expire cycle will try to avoid having more than
# ten percent of expired keys still in memory, and will try to avoid consuming
# more than 25% of total memory and to add latency to the system. However
# it is possible to increase the expire "effort" that is normally set to
# "1", to a greater value, up to the value "10". At its maximum value the
# system will use more CPU, longer cycles (and technically may introduce
# more latency), and will tollerate less already expired keys still present
# in the system. It's a tradeoff betweeen memory, CPU and latecy.
#
# active-expire-effort 1
############################# LAZY FREEING #################################### ############################# LAZY FREEING ####################################
# Redis has two primitives to delete keys. One is called DEL and is a blocking # Redis has two primitives to delete keys. One is called DEL and is a blocking
...@@ -1606,10 +1623,6 @@ rdb-save-incremental-fsync yes ...@@ -1606,10 +1623,6 @@ rdb-save-incremental-fsync yes
########################### ACTIVE DEFRAGMENTATION ####################### ########################### ACTIVE DEFRAGMENTATION #######################
# #
# WARNING THIS FEATURE IS EXPERIMENTAL. However it was stress tested
# even in production and manually tested by multiple engineers for some
# time.
#
# What is active defragmentation? # What is active defragmentation?
# ------------------------------- # -------------------------------
# #
...@@ -1649,7 +1662,7 @@ rdb-save-incremental-fsync yes ...@@ -1649,7 +1662,7 @@ rdb-save-incremental-fsync yes
# a good idea to leave the defaults untouched. # a good idea to leave the defaults untouched.
# Enabled active defragmentation # Enabled active defragmentation
# activedefrag yes # activedefrag no
# Minimum amount of fragmentation waste to start active defrag # Minimum amount of fragmentation waste to start active defrag
# active-defrag-ignore-bytes 100mb # active-defrag-ignore-bytes 100mb
...@@ -1660,11 +1673,13 @@ rdb-save-incremental-fsync yes ...@@ -1660,11 +1673,13 @@ rdb-save-incremental-fsync yes
# Maximum percentage of fragmentation at which we use maximum effort # Maximum percentage of fragmentation at which we use maximum effort
# active-defrag-threshold-upper 100 # active-defrag-threshold-upper 100
# Minimal effort for defrag in CPU percentage # Minimal effort for defrag in CPU percentage, to be used when the lower
# active-defrag-cycle-min 5 # threshold is reached
# active-defrag-cycle-min 1
# Maximal effort for defrag in CPU percentage # Maximal effort for defrag in CPU percentage, to be used when the upper
# active-defrag-cycle-max 75 # threshold is reached
# active-defrag-cycle-max 25
# Maximum number of set/hash/zset/list fields that will be processed from # Maximum number of set/hash/zset/list fields that will be processed from
# the main dictionary scan # the main dictionary scan
......
...@@ -22,4 +22,6 @@ $TCLSH tests/test_helper.tcl \ ...@@ -22,4 +22,6 @@ $TCLSH tests/test_helper.tcl \
--single unit/moduleapi/hooks \ --single unit/moduleapi/hooks \
--single unit/moduleapi/misc \ --single unit/moduleapi/misc \
--single unit/moduleapi/blockonkeys \ --single unit/moduleapi/blockonkeys \
--single unit/moduleapi/scan \
--single unit/moduleapi/datatype \
"${@}" "${@}"
...@@ -66,7 +66,7 @@ typedef struct list { ...@@ -66,7 +66,7 @@ typedef struct list {
#define listSetMatchMethod(l,m) ((l)->match = (m)) #define listSetMatchMethod(l,m) ((l)->match = (m))
#define listGetDupMethod(l) ((l)->dup) #define listGetDupMethod(l) ((l)->dup)
#define listGetFree(l) ((l)->free) #define listGetFreeMethod(l) ((l)->free)
#define listGetMatchMethod(l) ((l)->match) #define listGetMatchMethod(l) ((l)->match)
/* Prototypes */ /* Prototypes */
......
...@@ -1766,7 +1766,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) { ...@@ -1766,7 +1766,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
server.aof_selected_db = -1; /* Make sure SELECT is re-issued */ server.aof_selected_db = -1; /* Make sure SELECT is re-issued */
aofUpdateCurrentSize(); aofUpdateCurrentSize();
server.aof_rewrite_base_size = server.aof_current_size; server.aof_rewrite_base_size = server.aof_current_size;
server.aof_current_size = server.aof_current_size; server.aof_fsync_offset = server.aof_current_size;
/* Clear regular AOF buffer since its contents was just written to /* Clear regular AOF buffer since its contents was just written to
* the new AOF from the background rewrite buffer. */ * the new AOF from the background rewrite buffer. */
......
...@@ -514,6 +514,16 @@ void handleClientsBlockedOnKeys(void) { ...@@ -514,6 +514,16 @@ void handleClientsBlockedOnKeys(void) {
* we can safely call signalKeyAsReady() against this key. */ * we can safely call signalKeyAsReady() against this key. */
dictDelete(rl->db->ready_keys,rl->key); dictDelete(rl->db->ready_keys,rl->key);
/* Even if we are not inside call(), increment the call depth
* in order to make sure that keys are expired against a fixed
* reference time, and not against the wallclock time. This
* way we can lookup an object multiple times (BRPOPLPUSH does
* that) without the risk of it being freed in the second
* lookup, invalidating the first one.
* See https://github.com/antirez/redis/pull/6554. */
server.fixed_time_expire++;
updateCachedTime(0);
/* Serve clients blocked on list key. */ /* Serve clients blocked on list key. */
robj *o = lookupKeyWrite(rl->db,rl->key); robj *o = lookupKeyWrite(rl->db,rl->key);
...@@ -529,6 +539,7 @@ void handleClientsBlockedOnKeys(void) { ...@@ -529,6 +539,7 @@ void handleClientsBlockedOnKeys(void) {
* module is trying to accomplish right now. */ * module is trying to accomplish right now. */
serveClientsBlockedOnKeyByModule(rl); serveClientsBlockedOnKeyByModule(rl);
} }
server.fixed_time_expire--;
/* Free this item. */ /* Free this item. */
decrRefCount(rl->key); decrRefCount(rl->key);
......
...@@ -4966,7 +4966,7 @@ void restoreCommand(client *c) { ...@@ -4966,7 +4966,7 @@ void restoreCommand(client *c) {
if (!absttl) ttl+=mstime(); if (!absttl) ttl+=mstime();
setExpire(c,c->db,c->argv[1],ttl); setExpire(c,c->db,c->argv[1],ttl);
} }
objectSetLRUOrLFU(obj,lfu_freq,lru_idle,lru_clock); objectSetLRUOrLFU(obj,lfu_freq,lru_idle,lru_clock,1000);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
addReply(c,shared.ok); addReply(c,shared.ok);
server.dirty++; server.dirty++;
......
...@@ -256,7 +256,7 @@ void loadServerConfigFromString(char *config) { ...@@ -256,7 +256,7 @@ void loadServerConfigFromString(char *config) {
for (configYesNo *config = configs_yesno; config->name != NULL; config++) { for (configYesNo *config = configs_yesno; config->name != NULL; config++) {
if ((!strcasecmp(argv[0],config->name) || if ((!strcasecmp(argv[0],config->name) ||
(config->alias && !strcasecmp(argv[0],config->alias))) && (config->alias && !strcasecmp(argv[0],config->alias))) &&
(argc == 2)) (argc == 2))
{ {
if ((*(config->config) = yesnotoi(argv[1])) == -1) { if ((*(config->config) = yesnotoi(argv[1])) == -1) {
err = "argument must be 'yes' or 'no'"; goto loaderr; err = "argument must be 'yes' or 'no'"; goto loaderr;
...@@ -580,6 +580,14 @@ void loadServerConfigFromString(char *config) { ...@@ -580,6 +580,14 @@ void loadServerConfigFromString(char *config) {
err = "active-defrag-max-scan-fields must be positive"; err = "active-defrag-max-scan-fields must be positive";
goto loaderr; goto loaderr;
} }
} else if (!strcasecmp(argv[0],"active-expire-effort") && argc == 2) {
server.active_expire_effort = atoi(argv[1]);
if (server.active_expire_effort < 1 ||
server.active_expire_effort > 10)
{
err = "active-expire-effort must be between 1 and 10";
goto loaderr;
}
} else if (!strcasecmp(argv[0],"hash-max-ziplist-entries") && argc == 2) { } else if (!strcasecmp(argv[0],"hash-max-ziplist-entries") && argc == 2) {
server.hash_max_ziplist_entries = memtoll(argv[1], NULL); server.hash_max_ziplist_entries = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"hash-max-ziplist-value") && argc == 2) { } else if (!strcasecmp(argv[0],"hash-max-ziplist-value") && argc == 2) {
...@@ -1165,6 +1173,8 @@ void configSetCommand(client *c) { ...@@ -1165,6 +1173,8 @@ void configSetCommand(client *c) {
"active-defrag-cycle-max",server.active_defrag_cycle_max,1,99) { "active-defrag-cycle-max",server.active_defrag_cycle_max,1,99) {
} config_set_numerical_field( } config_set_numerical_field(
"active-defrag-max-scan-fields",server.active_defrag_max_scan_fields,1,LONG_MAX) { "active-defrag-max-scan-fields",server.active_defrag_max_scan_fields,1,LONG_MAX) {
} config_set_numerical_field(
"active-expire-effort",server.active_expire_effort,1,10) {
} config_set_numerical_field( } config_set_numerical_field(
"auto-aof-rewrite-percentage",server.aof_rewrite_perc,0,INT_MAX){ "auto-aof-rewrite-percentage",server.aof_rewrite_perc,0,INT_MAX){
} config_set_numerical_field( } config_set_numerical_field(
...@@ -1478,6 +1488,7 @@ void configGetCommand(client *c) { ...@@ -1478,6 +1488,7 @@ void configGetCommand(client *c) {
config_get_numerical_field("active-defrag-cycle-min",server.active_defrag_cycle_min); config_get_numerical_field("active-defrag-cycle-min",server.active_defrag_cycle_min);
config_get_numerical_field("active-defrag-cycle-max",server.active_defrag_cycle_max); config_get_numerical_field("active-defrag-cycle-max",server.active_defrag_cycle_max);
config_get_numerical_field("active-defrag-max-scan-fields",server.active_defrag_max_scan_fields); config_get_numerical_field("active-defrag-max-scan-fields",server.active_defrag_max_scan_fields);
config_get_numerical_field("active-expire-effort",server.active_expire_effort);
config_get_numerical_field("auto-aof-rewrite-percentage", config_get_numerical_field("auto-aof-rewrite-percentage",
server.aof_rewrite_perc); server.aof_rewrite_perc);
config_get_numerical_field("auto-aof-rewrite-min-size", config_get_numerical_field("auto-aof-rewrite-min-size",
...@@ -2327,6 +2338,7 @@ int rewriteConfig(char *path) { ...@@ -2327,6 +2338,7 @@ int rewriteConfig(char *path) {
rewriteConfigNumericalOption(state,"active-defrag-cycle-min",server.active_defrag_cycle_min,CONFIG_DEFAULT_DEFRAG_CYCLE_MIN); rewriteConfigNumericalOption(state,"active-defrag-cycle-min",server.active_defrag_cycle_min,CONFIG_DEFAULT_DEFRAG_CYCLE_MIN);
rewriteConfigNumericalOption(state,"active-defrag-cycle-max",server.active_defrag_cycle_max,CONFIG_DEFAULT_DEFRAG_CYCLE_MAX); rewriteConfigNumericalOption(state,"active-defrag-cycle-max",server.active_defrag_cycle_max,CONFIG_DEFAULT_DEFRAG_CYCLE_MAX);
rewriteConfigNumericalOption(state,"active-defrag-max-scan-fields",server.active_defrag_max_scan_fields,CONFIG_DEFAULT_DEFRAG_MAX_SCAN_FIELDS); rewriteConfigNumericalOption(state,"active-defrag-max-scan-fields",server.active_defrag_max_scan_fields,CONFIG_DEFAULT_DEFRAG_MAX_SCAN_FIELDS);
rewriteConfigNumericalOption(state,"active-expire-effort",server.active_expire_effort,CONFIG_DEFAULT_ACTIVE_EXPIRE_EFFORT);
rewriteConfigYesNoOption(state,"appendonly",server.aof_enabled,0); rewriteConfigYesNoOption(state,"appendonly",server.aof_enabled,0);
rewriteConfigStringOption(state,"appendfilename",server.aof_filename,CONFIG_DEFAULT_AOF_FILENAME); rewriteConfigStringOption(state,"appendfilename",server.aof_filename,CONFIG_DEFAULT_AOF_FILENAME);
rewriteConfigEnumOption(state,"appendfsync",server.aof_fsync,aof_fsync_enum,CONFIG_DEFAULT_AOF_FSYNC); rewriteConfigEnumOption(state,"appendfsync",server.aof_fsync,aof_fsync_enum,CONFIG_DEFAULT_AOF_FSYNC);
......
...@@ -1077,10 +1077,12 @@ int dbSwapDatabases(long id1, long id2) { ...@@ -1077,10 +1077,12 @@ int dbSwapDatabases(long id1, long id2) {
db1->dict = db2->dict; db1->dict = db2->dict;
db1->expires = db2->expires; db1->expires = db2->expires;
db1->avg_ttl = db2->avg_ttl; db1->avg_ttl = db2->avg_ttl;
db1->expires_cursor = db2->expires_cursor;
db2->dict = aux.dict; db2->dict = aux.dict;
db2->expires = aux.expires; db2->expires = aux.expires;
db2->avg_ttl = aux.avg_ttl; db2->avg_ttl = aux.avg_ttl;
db2->expires_cursor = aux.expires_cursor;
/* Now we need to handle clients blocked on lists: as an effect /* Now we need to handle clients blocked on lists: as an effect
* of swapping the two DBs, a client that was waiting for list * of swapping the two DBs, a client that was waiting for list
...@@ -1196,6 +1198,7 @@ void propagateExpire(redisDb *db, robj *key, int lazy) { ...@@ -1196,6 +1198,7 @@ void propagateExpire(redisDb *db, robj *key, int lazy) {
/* Check if the key is expired. */ /* Check if the key is expired. */
int keyIsExpired(redisDb *db, robj *key) { int keyIsExpired(redisDb *db, robj *key) {
mstime_t when = getExpire(db,key); mstime_t when = getExpire(db,key);
mstime_t now;
if (when < 0) return 0; /* No expire for this key */ if (when < 0) return 0; /* No expire for this key */
...@@ -1207,8 +1210,26 @@ int keyIsExpired(redisDb *db, robj *key) { ...@@ -1207,8 +1210,26 @@ int keyIsExpired(redisDb *db, robj *key) {
* only the first time it is accessed and not in the middle of the * only the first time it is accessed and not in the middle of the
* script execution, making propagation to slaves / AOF consistent. * script execution, making propagation to slaves / AOF consistent.
* See issue #1525 on Github for more information. */ * See issue #1525 on Github for more information. */
mstime_t now = server.lua_caller ? server.lua_time_start : mstime(); if (server.lua_caller) {
now = server.lua_time_start;
}
/* If we are in the middle of a command execution, we still want to use
* a reference time that does not change: in that case we just use the
* cached time, that we update before each call in the call() function.
* This way we avoid that commands such as RPOPLPUSH or similar, that
* may re-open the same key multiple times, can invalidate an already
* open object in a next call, if the next call will see the key expired,
* while the first did not. */
else if (server.fixed_time_expire > 0) {
now = server.mstime;
}
/* For the other cases, we want to use the most fresh time we have. */
else {
now = mstime();
}
/* The key expired if the current (virtual or real) time is greater
* than the expire time of the key. */
return now > when; return now > when;
} }
......
...@@ -919,10 +919,12 @@ int defragLaterItem(dictEntry *de, unsigned long *cursor, long long endtime) { ...@@ -919,10 +919,12 @@ int defragLaterItem(dictEntry *de, unsigned long *cursor, long long endtime) {
return 0; return 0;
} }
/* static variables serving defragLaterStep to continue scanning a key from were we stopped last time. */
static sds defrag_later_current_key = NULL;
static unsigned long defrag_later_cursor = 0;
/* returns 0 if no more work needs to be been done, and 1 if time is up and more work is needed. */ /* returns 0 if no more work needs to be been done, and 1 if time is up and more work is needed. */
int defragLaterStep(redisDb *db, long long endtime) { int defragLaterStep(redisDb *db, long long endtime) {
static sds current_key = NULL;
static unsigned long cursor = 0;
unsigned int iterations = 0; unsigned int iterations = 0;
unsigned long long prev_defragged = server.stat_active_defrag_hits; unsigned long long prev_defragged = server.stat_active_defrag_hits;
unsigned long long prev_scanned = server.stat_active_defrag_scanned; unsigned long long prev_scanned = server.stat_active_defrag_scanned;
...@@ -930,16 +932,15 @@ int defragLaterStep(redisDb *db, long long endtime) { ...@@ -930,16 +932,15 @@ int defragLaterStep(redisDb *db, long long endtime) {
do { do {
/* if we're not continuing a scan from the last call or loop, start a new one */ /* if we're not continuing a scan from the last call or loop, start a new one */
if (!cursor) { if (!defrag_later_cursor) {
listNode *head = listFirst(db->defrag_later); listNode *head = listFirst(db->defrag_later);
/* Move on to next key */ /* Move on to next key */
if (current_key) { if (defrag_later_current_key) {
serverAssert(current_key == head->value); serverAssert(defrag_later_current_key == head->value);
sdsfree(head->value);
listDelNode(db->defrag_later, head); listDelNode(db->defrag_later, head);
cursor = 0; defrag_later_cursor = 0;
current_key = NULL; defrag_later_current_key = NULL;
} }
/* stop if we reached the last one. */ /* stop if we reached the last one. */
...@@ -948,21 +949,21 @@ int defragLaterStep(redisDb *db, long long endtime) { ...@@ -948,21 +949,21 @@ int defragLaterStep(redisDb *db, long long endtime) {
return 0; return 0;
/* start a new key */ /* start a new key */
current_key = head->value; defrag_later_current_key = head->value;
cursor = 0; defrag_later_cursor = 0;
} }
/* each time we enter this function we need to fetch the key from the dict again (if it still exists) */ /* each time we enter this function we need to fetch the key from the dict again (if it still exists) */
dictEntry *de = dictFind(db->dict, current_key); dictEntry *de = dictFind(db->dict, defrag_later_current_key);
key_defragged = server.stat_active_defrag_hits; key_defragged = server.stat_active_defrag_hits;
do { do {
int quit = 0; int quit = 0;
if (defragLaterItem(de, &cursor, endtime)) if (defragLaterItem(de, &defrag_later_cursor, endtime))
quit = 1; /* time is up, we didn't finish all the work */ quit = 1; /* time is up, we didn't finish all the work */
/* Don't start a new BIG key in this loop, this is because the /* Don't start a new BIG key in this loop, this is because the
* next key can be a list, and scanLaterList must be done in once cycle */ * next key can be a list, and scanLaterList must be done in once cycle */
if (!cursor) if (!defrag_later_cursor)
quit = 1; quit = 1;
/* Once in 16 scan iterations, 512 pointer reallocations, or 64 fields /* Once in 16 scan iterations, 512 pointer reallocations, or 64 fields
...@@ -982,7 +983,7 @@ int defragLaterStep(redisDb *db, long long endtime) { ...@@ -982,7 +983,7 @@ int defragLaterStep(redisDb *db, long long endtime) {
prev_defragged = server.stat_active_defrag_hits; prev_defragged = server.stat_active_defrag_hits;
prev_scanned = server.stat_active_defrag_scanned; prev_scanned = server.stat_active_defrag_scanned;
} }
} while(cursor); } while(defrag_later_cursor);
if(key_defragged != server.stat_active_defrag_hits) if(key_defragged != server.stat_active_defrag_hits)
server.stat_active_defrag_key_hits++; server.stat_active_defrag_key_hits++;
else else
...@@ -1039,6 +1040,21 @@ void activeDefragCycle(void) { ...@@ -1039,6 +1040,21 @@ void activeDefragCycle(void) {
mstime_t latency; mstime_t latency;
int quit = 0; int quit = 0;
if (!server.active_defrag_enabled) {
if (server.active_defrag_running) {
/* if active defrag was disabled mid-run, start from fresh next time. */
server.active_defrag_running = 0;
if (db)
listEmpty(db->defrag_later);
defrag_later_current_key = NULL;
defrag_later_cursor = 0;
current_db = -1;
cursor = 0;
db = NULL;
}
return;
}
if (hasActiveChildProcess()) if (hasActiveChildProcess())
return; /* Defragging memory while there's a fork will just do damage. */ return; /* Defragging memory while there's a fork will just do damage. */
......
...@@ -78,24 +78,63 @@ int activeExpireCycleTryExpire(redisDb *db, dictEntry *de, long long now) { ...@@ -78,24 +78,63 @@ int activeExpireCycleTryExpire(redisDb *db, dictEntry *de, long long now) {
* it will get more aggressive to avoid that too much memory is used by * it will get more aggressive to avoid that too much memory is used by
* keys that can be removed from the keyspace. * keys that can be removed from the keyspace.
* *
* No more than CRON_DBS_PER_CALL databases are tested at every * Every expire cycle tests multiple databases: the next call will start
* iteration. * again from the next db, with the exception of exists for time limit: in that
* case we restart again from the last database we were processing. Anyway
* no more than CRON_DBS_PER_CALL databases are tested at every iteration.
* *
* This kind of call is used when Redis detects that timelimit_exit is * The function can perform more or less work, depending on the "type"
* true, so there is more work to do, and we do it more incrementally from * argument. It can execute a "fast cycle" or a "slow cycle". The slow
* the beforeSleep() function of the event loop. * cycle is the main way we collect expired cycles: this happens with
* the "server.hz" frequency (usually 10 hertz).
* *
* Expire cycle type: * However the slow cycle can exit for timeout, since it used too much time.
* For this reason the function is also invoked to perform a fast cycle
* at every event loop cycle, in the beforeSleep() function. The fast cycle
* will try to perform less work, but will do it much more often.
*
* The following are the details of the two expire cycles and their stop
* conditions:
* *
* If type is ACTIVE_EXPIRE_CYCLE_FAST the function will try to run a * If type is ACTIVE_EXPIRE_CYCLE_FAST the function will try to run a
* "fast" expire cycle that takes no longer than EXPIRE_FAST_CYCLE_DURATION * "fast" expire cycle that takes no longer than EXPIRE_FAST_CYCLE_DURATION
* microseconds, and is not repeated again before the same amount of time. * microseconds, and is not repeated again before the same amount of time.
* The cycle will also refuse to run at all if the latest slow cycle did not
* terminate because of a time limit condition.
* *
* If type is ACTIVE_EXPIRE_CYCLE_SLOW, that normal expire cycle is * If type is ACTIVE_EXPIRE_CYCLE_SLOW, that normal expire cycle is
* executed, where the time limit is a percentage of the REDIS_HZ period * executed, where the time limit is a percentage of the REDIS_HZ period
* as specified by the ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC define. */ * as specified by the ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC define. In the
* fast cycle, the check of every database is interrupted once the number
* of already expired keys in the database is estimated to be lower than
* a given percentage, in order to avoid doing too much work to gain too
* little memory.
*
* The configured expire "effort" will modify the baseline parameters in
* order to do more work in both the fast and slow expire cycles.
*/
#define ACTIVE_EXPIRE_CYCLE_KEYS_PER_LOOP 20 /* Keys for each DB loop. */
#define ACTIVE_EXPIRE_CYCLE_FAST_DURATION 1000 /* Microseconds. */
#define ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 25 /* Max % of CPU to use. */
#define ACTIVE_EXPIRE_CYCLE_ACCEPTABLE_STALE 10 /* % of stale keys after which
we do extra efforts. */
void activeExpireCycle(int type) { void activeExpireCycle(int type) {
/* Adjust the running parameters according to the configured expire
* effort. The default effort is 1, and the maximum configurable effort
* is 10. */
unsigned long
effort = server.active_expire_effort-1, /* Rescale from 0 to 9. */
config_keys_per_loop = ACTIVE_EXPIRE_CYCLE_KEYS_PER_LOOP +
ACTIVE_EXPIRE_CYCLE_KEYS_PER_LOOP/4*effort,
config_cycle_fast_duration = ACTIVE_EXPIRE_CYCLE_FAST_DURATION +
ACTIVE_EXPIRE_CYCLE_FAST_DURATION/4*effort,
config_cycle_slow_time_perc = ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC +
2*effort,
config_cycle_acceptable_stale = ACTIVE_EXPIRE_CYCLE_ACCEPTABLE_STALE-
effort;
/* This function has some global state in order to continue the work /* This function has some global state in order to continue the work
* incrementally across calls. */ * incrementally across calls. */
static unsigned int current_db = 0; /* Last DB tested. */ static unsigned int current_db = 0; /* Last DB tested. */
...@@ -113,10 +152,16 @@ void activeExpireCycle(int type) { ...@@ -113,10 +152,16 @@ void activeExpireCycle(int type) {
if (type == ACTIVE_EXPIRE_CYCLE_FAST) { if (type == ACTIVE_EXPIRE_CYCLE_FAST) {
/* Don't start a fast cycle if the previous cycle did not exit /* Don't start a fast cycle if the previous cycle did not exit
* for time limit. Also don't repeat a fast cycle for the same period * for time limit, unless the percentage of estimated stale keys is
* too high. Also never repeat a fast cycle for the same period
* as the fast cycle total duration itself. */ * as the fast cycle total duration itself. */
if (!timelimit_exit) return; if (!timelimit_exit &&
if (start < last_fast_cycle + ACTIVE_EXPIRE_CYCLE_FAST_DURATION*2) return; server.stat_expired_stale_perc < config_cycle_acceptable_stale)
return;
if (start < last_fast_cycle + (long long)config_cycle_fast_duration*2)
return;
last_fast_cycle = start; last_fast_cycle = start;
} }
...@@ -130,16 +175,16 @@ void activeExpireCycle(int type) { ...@@ -130,16 +175,16 @@ void activeExpireCycle(int type) {
if (dbs_per_call > server.dbnum || timelimit_exit) if (dbs_per_call > server.dbnum || timelimit_exit)
dbs_per_call = server.dbnum; dbs_per_call = server.dbnum;
/* We can use at max ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC percentage of CPU time /* We can use at max 'config_cycle_slow_time_perc' percentage of CPU
* per iteration. Since this function gets called with a frequency of * time per iteration. Since this function gets called with a frequency of
* server.hz times per second, the following is the max amount of * server.hz times per second, the following is the max amount of
* microseconds we can spend in this function. */ * microseconds we can spend in this function. */
timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100; timelimit = config_cycle_slow_time_perc*1000000/server.hz/100;
timelimit_exit = 0; timelimit_exit = 0;
if (timelimit <= 0) timelimit = 1; if (timelimit <= 0) timelimit = 1;
if (type == ACTIVE_EXPIRE_CYCLE_FAST) if (type == ACTIVE_EXPIRE_CYCLE_FAST)
timelimit = ACTIVE_EXPIRE_CYCLE_FAST_DURATION; /* in microseconds. */ timelimit = config_cycle_fast_duration; /* in microseconds. */
/* Accumulate some global stats as we expire keys, to have some idea /* Accumulate some global stats as we expire keys, to have some idea
* about the number of keys that are already logically expired, but still * about the number of keys that are already logically expired, but still
...@@ -148,7 +193,9 @@ void activeExpireCycle(int type) { ...@@ -148,7 +193,9 @@ void activeExpireCycle(int type) {
long total_expired = 0; long total_expired = 0;
for (j = 0; j < dbs_per_call && timelimit_exit == 0; j++) { for (j = 0; j < dbs_per_call && timelimit_exit == 0; j++) {
int expired; /* Expired and checked in a single loop. */
unsigned long expired, sampled;
redisDb *db = server.db+(current_db % server.dbnum); redisDb *db = server.db+(current_db % server.dbnum);
/* Increment the DB now so we are sure if we run out of time /* Increment the DB now so we are sure if we run out of time
...@@ -172,8 +219,8 @@ void activeExpireCycle(int type) { ...@@ -172,8 +219,8 @@ void activeExpireCycle(int type) {
slots = dictSlots(db->expires); slots = dictSlots(db->expires);
now = mstime(); now = mstime();
/* When there are less than 1% filled slots getting random /* When there are less than 1% filled slots, sampling the key
* keys is expensive, so stop here waiting for better times... * space is expensive, so stop here waiting for better times...
* The dictionary will be resized asap. */ * The dictionary will be resized asap. */
if (num && slots > DICT_HT_INITIAL_SIZE && if (num && slots > DICT_HT_INITIAL_SIZE &&
(num*100/slots < 1)) break; (num*100/slots < 1)) break;
...@@ -181,27 +228,58 @@ void activeExpireCycle(int type) { ...@@ -181,27 +228,58 @@ void activeExpireCycle(int type) {
/* The main collection cycle. Sample random keys among keys /* The main collection cycle. Sample random keys among keys
* with an expire set, checking for expired ones. */ * with an expire set, checking for expired ones. */
expired = 0; expired = 0;
sampled = 0;
ttl_sum = 0; ttl_sum = 0;
ttl_samples = 0; ttl_samples = 0;
if (num > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP) if (num > config_keys_per_loop)
num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP; num = config_keys_per_loop;
while (num--) { /* Here we access the low level representation of the hash table
dictEntry *de; * for speed concerns: this makes this code coupled with dict.c,
long long ttl; * but it hardly changed in ten years.
*
if ((de = dictGetRandomKey(db->expires)) == NULL) break; * Note that certain places of the hash table may be empty,
ttl = dictGetSignedIntegerVal(de)-now; * so we want also a stop condition about the number of
if (activeExpireCycleTryExpire(db,de,now)) expired++; * buckets that we scanned. However scanning for free buckets
if (ttl > 0) { * is very fast: we are in the cache line scanning a sequential
/* We want the average TTL of keys yet not expired. */ * array of NULL pointers, so we can scan a lot more buckets
ttl_sum += ttl; * than keys in the same time. */
ttl_samples++; long max_buckets = num*20;
long checked_buckets = 0;
while (sampled < num && checked_buckets < max_buckets) {
for (int table = 0; table < 2; table++) {
if (table == 1 && !dictIsRehashing(db->expires)) break;
unsigned long idx = db->expires_cursor;
idx &= db->expires->ht[table].sizemask;
dictEntry *de = db->expires->ht[table].table[idx];
long long ttl;
/* Scan the current bucket of the current table. */
checked_buckets++;
while(de) {
/* Get the next entry now since this entry may get
* deleted. */
dictEntry *e = de;
de = de->next;
ttl = dictGetSignedIntegerVal(e)-now;
if (activeExpireCycleTryExpire(db,e,now)) expired++;
if (ttl > 0) {
/* We want the average TTL of keys yet
* not expired. */
ttl_sum += ttl;
ttl_samples++;
}
sampled++;
}
} }
total_sampled++; db->expires_cursor++;
} }
total_expired += expired; total_expired += expired;
total_sampled += sampled;
/* Update the average TTL stats for this database. */ /* Update the average TTL stats for this database. */
if (ttl_samples) { if (ttl_samples) {
...@@ -225,12 +303,14 @@ void activeExpireCycle(int type) { ...@@ -225,12 +303,14 @@ void activeExpireCycle(int type) {
break; break;
} }
} }
/* We don't repeat the cycle if there are less than 25% of keys /* We don't repeat the cycle for the current database if there are
* found expired in the current DB. */ * an acceptable amount of stale keys (logically expired but yet
} while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4); * not reclained). */
} while ((expired*100/sampled) > config_cycle_acceptable_stale);
} }
elapsed = ustime()-start; elapsed = ustime()-start;
server.stat_expire_cycle_time_used += elapsed;
latencyAddSampleIfNeeded("expire-cycle",elapsed/1000); latencyAddSampleIfNeeded("expire-cycle",elapsed/1000);
/* Update our estimate of keys existing but yet to be expired. /* Update our estimate of keys existing but yet to be expired.
......
...@@ -3,7 +3,7 @@ GIT_SHA1=`(git show-ref --head --hash=8 2> /dev/null || echo 00000000) | head -n ...@@ -3,7 +3,7 @@ GIT_SHA1=`(git show-ref --head --hash=8 2> /dev/null || echo 00000000) | head -n
GIT_DIRTY=`git diff --no-ext-diff 2> /dev/null | wc -l` GIT_DIRTY=`git diff --no-ext-diff 2> /dev/null | wc -l`
BUILD_ID=`uname -n`"-"`date +%s` BUILD_ID=`uname -n`"-"`date +%s`
if [ -n "$SOURCE_DATE_EPOCH" ]; then if [ -n "$SOURCE_DATE_EPOCH" ]; then
BUILD_ID=$(date -u -d "@$SOURCE_DATE_EPOCH" +%s 2>/dev/null || date -u -r "$SOURCE_DATE_EPOCH" +%s 2>/dev/null || date -u %s) BUILD_ID=$(date -u -d "@$SOURCE_DATE_EPOCH" +%s 2>/dev/null || date -u -r "$SOURCE_DATE_EPOCH" +%s 2>/dev/null || date -u +%s)
fi fi
test -f release.h || touch release.h test -f release.h || touch release.h
(cat release.h | grep SHA1 | grep $GIT_SHA1) && \ (cat release.h | grep SHA1 | grep $GIT_SHA1) && \
......
...@@ -1012,6 +1012,21 @@ RedisModuleString *RM_CreateStringFromLongLong(RedisModuleCtx *ctx, long long ll ...@@ -1012,6 +1012,21 @@ RedisModuleString *RM_CreateStringFromLongLong(RedisModuleCtx *ctx, long long ll
return RM_CreateString(ctx,buf,len); return RM_CreateString(ctx,buf,len);
} }
/* Like RedisModule_CreatString(), but creates a string starting from a long
* double.
*
* The returned string must be released with RedisModule_FreeString() or by
* enabling automatic memory management.
*
* The passed context 'ctx' may be NULL if necessary, see the
* RedisModule_CreateString() documentation for more info. */
RedisModuleString *RM_CreateStringFromLongDouble(RedisModuleCtx *ctx, long double ld, int humanfriendly) {
char buf[MAX_LONG_DOUBLE_CHARS];
size_t len = ld2string(buf,sizeof(buf),ld,
(humanfriendly ? LD_STR_HUMAN : LD_STR_AUTO));
return RM_CreateString(ctx,buf,len);
}
/* Like RedisModule_CreatString(), but creates a string starting from another /* Like RedisModule_CreatString(), but creates a string starting from another
* RedisModuleString. * RedisModuleString.
* *
...@@ -1116,6 +1131,14 @@ int RM_StringToDouble(const RedisModuleString *str, double *d) { ...@@ -1116,6 +1131,14 @@ int RM_StringToDouble(const RedisModuleString *str, double *d) {
return (retval == C_OK) ? REDISMODULE_OK : REDISMODULE_ERR; return (retval == C_OK) ? REDISMODULE_OK : REDISMODULE_ERR;
} }
/* Convert the string into a long double, storing it at `*ld`.
* Returns REDISMODULE_OK on success or REDISMODULE_ERR if the string is
* not a valid string representation of a double value. */
int RM_StringToLongDouble(const RedisModuleString *str, long double *ld) {
int retval = string2ld(str->ptr,sdslen(str->ptr),ld);
return retval ? REDISMODULE_OK : REDISMODULE_ERR;
}
/* Compare two string objects, returning -1, 0 or 1 respectively if /* Compare two string objects, returning -1, 0 or 1 respectively if
* a < b, a == b, a > b. Strings are compared byte by byte as two * a < b, a == b, a > b. Strings are compared byte by byte as two
* binary blobs without any encoding care / collation attempt. */ * binary blobs without any encoding care / collation attempt. */
...@@ -1389,7 +1412,7 @@ int RM_ReplyWithString(RedisModuleCtx *ctx, RedisModuleString *str) { ...@@ -1389,7 +1412,7 @@ int RM_ReplyWithString(RedisModuleCtx *ctx, RedisModuleString *str) {
int RM_ReplyWithEmptyString(RedisModuleCtx *ctx) { int RM_ReplyWithEmptyString(RedisModuleCtx *ctx) {
client *c = moduleGetReplyClient(ctx); client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK; if (c == NULL) return REDISMODULE_OK;
addReplyBulkCBuffer(c, "", 0); addReply(c,shared.emptybulk);
return REDISMODULE_OK; return REDISMODULE_OK;
} }
...@@ -1404,8 +1427,7 @@ int RM_ReplyWithVerbatimString(RedisModuleCtx *ctx, const char *buf, size_t len) ...@@ -1404,8 +1427,7 @@ int RM_ReplyWithVerbatimString(RedisModuleCtx *ctx, const char *buf, size_t len)
return REDISMODULE_OK; return REDISMODULE_OK;
} }
/* Reply to the client with a NULL. In the RESP protocol a NULL is encoded /* Reply to the client with a NULL.
* as the string "$-1\r\n".
* *
* The function always returns REDISMODULE_OK. */ * The function always returns REDISMODULE_OK. */
int RM_ReplyWithNull(RedisModuleCtx *ctx) { int RM_ReplyWithNull(RedisModuleCtx *ctx) {
...@@ -1442,6 +1464,21 @@ int RM_ReplyWithDouble(RedisModuleCtx *ctx, double d) { ...@@ -1442,6 +1464,21 @@ int RM_ReplyWithDouble(RedisModuleCtx *ctx, double d) {
return REDISMODULE_OK; return REDISMODULE_OK;
} }
/* Send a string reply obtained converting the long double 'ld' into a bulk
* string. This function is basically equivalent to converting a long double
* into a string into a C buffer, and then calling the function
* RedisModule_ReplyWithStringBuffer() with the buffer and length.
* The double string uses human readable formatting (see
* `addReplyHumanLongDouble` in networking.c).
*
* The function always returns REDISMODULE_OK. */
int RM_ReplyWithLongDouble(RedisModuleCtx *ctx, long double ld) {
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
addReplyHumanLongDouble(c, ld);
return REDISMODULE_OK;
}
/* -------------------------------------------------------------------------- /* --------------------------------------------------------------------------
* Commands replication API * Commands replication API
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
...@@ -1749,6 +1786,8 @@ int RM_GetSelectedDb(RedisModuleCtx *ctx) { ...@@ -1749,6 +1786,8 @@ int RM_GetSelectedDb(RedisModuleCtx *ctx) {
* * REDISMODULE_CTX_FLAGS_OOM_WARNING: Less than 25% of memory remains before * * REDISMODULE_CTX_FLAGS_OOM_WARNING: Less than 25% of memory remains before
* reaching the maxmemory level. * reaching the maxmemory level.
* *
* * REDISMODULE_CTX_FLAGS_LOADING: Server is loading RDB/AOF
*
* * REDISMODULE_CTX_FLAGS_REPLICA_IS_STALE: No active link with the master. * * REDISMODULE_CTX_FLAGS_REPLICA_IS_STALE: No active link with the master.
* *
* * REDISMODULE_CTX_FLAGS_REPLICA_IS_CONNECTING: The replica is trying to * * REDISMODULE_CTX_FLAGS_REPLICA_IS_CONNECTING: The replica is trying to
...@@ -1848,6 +1887,18 @@ int RM_SelectDb(RedisModuleCtx *ctx, int newid) { ...@@ -1848,6 +1887,18 @@ int RM_SelectDb(RedisModuleCtx *ctx, int newid) {
return (retval == C_OK) ? REDISMODULE_OK : REDISMODULE_ERR; return (retval == C_OK) ? REDISMODULE_OK : REDISMODULE_ERR;
} }
/* Initialize a RedisModuleKey struct */
static void moduleInitKey(RedisModuleKey *kp, RedisModuleCtx *ctx, robj *keyname, robj *value, int mode){
kp->ctx = ctx;
kp->db = ctx->client->db;
kp->key = keyname;
incrRefCount(keyname);
kp->value = value;
kp->iter = NULL;
kp->mode = mode;
zsetKeyReset(kp);
}
/* Return an handle representing a Redis key, so that it is possible /* Return an handle representing a Redis key, so that it is possible
* to call other APIs with the key handle as argument to perform * to call other APIs with the key handle as argument to perform
* operations on the key. * operations on the key.
...@@ -1878,27 +1929,25 @@ void *RM_OpenKey(RedisModuleCtx *ctx, robj *keyname, int mode) { ...@@ -1878,27 +1929,25 @@ void *RM_OpenKey(RedisModuleCtx *ctx, robj *keyname, int mode) {
/* Setup the key handle. */ /* Setup the key handle. */
kp = zmalloc(sizeof(*kp)); kp = zmalloc(sizeof(*kp));
kp->ctx = ctx; moduleInitKey(kp, ctx, keyname, value, mode);
kp->db = ctx->client->db;
kp->key = keyname;
incrRefCount(keyname);
kp->value = value;
kp->iter = NULL;
kp->mode = mode;
zsetKeyReset(kp);
autoMemoryAdd(ctx,REDISMODULE_AM_KEY,kp); autoMemoryAdd(ctx,REDISMODULE_AM_KEY,kp);
return (void*)kp; return (void*)kp;
} }
/* Close a key handle. */ /* Destroy a RedisModuleKey struct (freeing is the responsibility of the caller). */
void RM_CloseKey(RedisModuleKey *key) { static void moduleCloseKey(RedisModuleKey *key) {
if (key == NULL) return;
int signal = SHOULD_SIGNAL_MODIFIED_KEYS(key->ctx); int signal = SHOULD_SIGNAL_MODIFIED_KEYS(key->ctx);
if ((key->mode & REDISMODULE_WRITE) && signal) if ((key->mode & REDISMODULE_WRITE) && signal)
signalModifiedKey(key->db,key->key); signalModifiedKey(key->db,key->key);
/* TODO: if (key->iter) RM_KeyIteratorStop(kp); */ /* TODO: if (key->iter) RM_KeyIteratorStop(kp); */
RM_ZsetRangeStop(key); RM_ZsetRangeStop(key);
decrRefCount(key->key); decrRefCount(key->key);
}
/* Close a key handle. */
void RM_CloseKey(RedisModuleKey *key) {
if (key == NULL) return;
moduleCloseKey(key);
autoMemoryFreed(key->ctx,REDISMODULE_AM_KEY,key); autoMemoryFreed(key->ctx,REDISMODULE_AM_KEY,key);
zfree(key); zfree(key);
} }
...@@ -1950,7 +1999,7 @@ int RM_DeleteKey(RedisModuleKey *key) { ...@@ -1950,7 +1999,7 @@ int RM_DeleteKey(RedisModuleKey *key) {
return REDISMODULE_OK; return REDISMODULE_OK;
} }
/* If the key is open for writing, unlink it (that is delete it in a /* If the key is open for writing, unlink it (that is delete it in a
* non-blocking way, not reclaiming memory immediately) and setup the key to * non-blocking way, not reclaiming memory immediately) and setup the key to
* accept new writes as an empty key (that will be created on demand). * accept new writes as an empty key (that will be created on demand).
* On success REDISMODULE_OK is returned. If the key is not open for * On success REDISMODULE_OK is returned. If the key is not open for
...@@ -3110,7 +3159,9 @@ fmterr: ...@@ -3110,7 +3159,9 @@ fmterr:
* On success a RedisModuleCallReply object is returned, otherwise * On success a RedisModuleCallReply object is returned, otherwise
* NULL is returned and errno is set to the following values: * NULL is returned and errno is set to the following values:
* *
* EINVAL: command non existing, wrong arity, wrong format specifier. * EBADF: wrong format specifier.
* EINVAL: wrong command arity.
* ENOENT: command does not exist.
* EPERM: operation in Cluster instance with key in non local slot. * EPERM: operation in Cluster instance with key in non local slot.
* *
* This API is documented here: https://redis.io/topics/modules-intro * This API is documented here: https://redis.io/topics/modules-intro
...@@ -3142,7 +3193,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch ...@@ -3142,7 +3193,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
/* We handle the above format error only when the client is setup so that /* We handle the above format error only when the client is setup so that
* we can free it normally. */ * we can free it normally. */
if (argv == NULL) { if (argv == NULL) {
errno = EINVAL; errno = EBADF;
goto cleanup; goto cleanup;
} }
...@@ -3154,7 +3205,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch ...@@ -3154,7 +3205,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
*/ */
cmd = lookupCommand(c->argv[0]->ptr); cmd = lookupCommand(c->argv[0]->ptr);
if (!cmd) { if (!cmd) {
errno = EINVAL; errno = ENOENT;
goto cleanup; goto cleanup;
} }
c->cmd = c->lastcmd = cmd; c->cmd = c->lastcmd = cmd;
...@@ -3884,6 +3935,59 @@ void RM_DigestEndSequence(RedisModuleDigest *md) { ...@@ -3884,6 +3935,59 @@ void RM_DigestEndSequence(RedisModuleDigest *md) {
memset(md->o,0,sizeof(md->o)); memset(md->o,0,sizeof(md->o));
} }
/* Decode a serialized representation of a module data type 'mt' from string
* 'str' and return a newly allocated value, or NULL if decoding failed.
*
* This call basically reuses the 'rdb_load' callback which module data types
* implement in order to allow a module to arbitrarily serialize/de-serialize
* keys, similar to how the Redis 'DUMP' and 'RESTORE' commands are implemented.
*
* Modules should generally use the REDISMODULE_OPTIONS_HANDLE_IO_ERRORS flag and
* make sure the de-serialization code properly checks and handles IO errors
* (freeing allocated buffers and returning a NULL).
*
* If this is NOT done, Redis will handle corrupted (or just truncated) serialized
* data by producing an error message and terminating the process.
*/
void *RM_LoadDataTypeFromString(const RedisModuleString *str, const moduleType *mt) {
rio payload;
RedisModuleIO io;
rioInitWithBuffer(&payload, str->ptr);
moduleInitIOContext(io,(moduleType *)mt,&payload,NULL);
/* All RM_Save*() calls always write a version 2 compatible format, so we
* need to make sure we read the same.
*/
io.ver = 2;
return mt->rdb_load(&io,0);
}
/* Encode a module data type 'mt' value 'data' into serialized form, and return it
* as a newly allocated RedisModuleString.
*
* This call basically reuses the 'rdb_save' callback which module data types
* implement in order to allow a module to arbitrarily serialize/de-serialize
* keys, similar to how the Redis 'DUMP' and 'RESTORE' commands are implemented.
*/
RedisModuleString *RM_SaveDataTypeToString(RedisModuleCtx *ctx, void *data, const moduleType *mt) {
rio payload;
RedisModuleIO io;
rioInitWithBuffer(&payload,sdsempty());
moduleInitIOContext(io,(moduleType *)mt,&payload,NULL);
mt->rdb_save(&io,data);
if (io.error) {
return NULL;
} else {
robj *str = createObject(OBJ_STRING,payload.io.buffer.ptr);
autoMemoryAdd(ctx,REDISMODULE_AM_STRING,str);
return str;
}
}
/* -------------------------------------------------------------------------- /* --------------------------------------------------------------------------
* AOF API for modules data types * AOF API for modules data types
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
...@@ -5911,6 +6015,239 @@ float RM_GetUsedMemoryRatio(){ ...@@ -5911,6 +6015,239 @@ float RM_GetUsedMemoryRatio(){
return level; return level;
} }
/* --------------------------------------------------------------------------
* Scanning keyspace and hashes
* -------------------------------------------------------------------------- */
typedef void (*RedisModuleScanCB)(RedisModuleCtx *ctx, RedisModuleString *keyname, RedisModuleKey *key, void *privdata);
typedef struct {
RedisModuleCtx *ctx;
void* user_data;
RedisModuleScanCB fn;
} ScanCBData;
typedef struct RedisModuleScanCursor{
int cursor;
int done;
}RedisModuleScanCursor;
static void moduleScanCallback(void *privdata, const dictEntry *de) {
ScanCBData *data = privdata;
sds key = dictGetKey(de);
robj* val = dictGetVal(de);
RedisModuleString *keyname = createObject(OBJ_STRING,sdsdup(key));
/* Setup the key handle. */
RedisModuleKey kp = {0};
moduleInitKey(&kp, data->ctx, keyname, val, REDISMODULE_READ);
data->fn(data->ctx, keyname, &kp, data->user_data);
moduleCloseKey(&kp);
decrRefCount(keyname);
}
/* Create a new cursor to be used with RedisModule_Scan */
RedisModuleScanCursor *RM_ScanCursorCreate() {
RedisModuleScanCursor* cursor = zmalloc(sizeof(*cursor));
cursor->cursor = 0;
cursor->done = 0;
return cursor;
}
/* Restart an existing cursor. The keys will be rescanned. */
void RM_ScanCursorRestart(RedisModuleScanCursor *cursor) {
cursor->cursor = 0;
cursor->done = 0;
}
/* Destroy the cursor struct. */
void RM_ScanCursorDestroy(RedisModuleScanCursor *cursor) {
zfree(cursor);
}
/* Scan api that allows a module to scan all the keys and value in the selected db.
*
* Callback for scan implementation.
* void scan_callback(RedisModuleCtx *ctx, RedisModuleString *keyname, RedisModuleKey *key, void *privdata);
* - ctx - the redis module context provided to for the scan.
* - keyname - owned by the caller and need to be retained if used after this function.
* - key - holds info on the key and value, it is provided as best effort, in some cases it might
* be NULL, in which case the user should (can) use RedisModule_OpenKey (and CloseKey too).
* when it is provided, it is owned by the caller and will be free when the callback returns.
* - privdata - the user data provided to RedisModule_Scan.
*
* The way it should be used:
* RedisModuleCursor *c = RedisModule_ScanCursorCreate();
* while(RedisModule_Scan(ctx, c, callback, privateData));
* RedisModule_ScanCursorDestroy(c);
*
* It is also possible to use this API from another thread while the lock is acquired durring
* the actuall call to RM_Scan:
* RedisModuleCursor *c = RedisModule_ScanCursorCreate();
* RedisModule_ThreadSafeContextLock(ctx);
* while(RedisModule_Scan(ctx, c, callback, privateData)){
* RedisModule_ThreadSafeContextUnlock(ctx);
* // do some background job
* RedisModule_ThreadSafeContextLock(ctx);
* }
* RedisModule_ScanCursorDestroy(c);
*
* The function will return 1 if there are more elements to scan and 0 otherwise,
* possibly setting errno if the call failed.
* It is also possible to restart and existing cursor using RM_CursorRestart. */
int RM_Scan(RedisModuleCtx *ctx, RedisModuleScanCursor *cursor, RedisModuleScanCB fn, void *privdata) {
if (cursor->done) {
errno = ENOENT;
return 0;
}
int ret = 1;
ScanCBData data = { ctx, privdata, fn };
cursor->cursor = dictScan(ctx->client->db->dict, cursor->cursor, moduleScanCallback, NULL, &data);
if (cursor->cursor == 0) {
cursor->done = 1;
ret = 0;
}
errno = 0;
return ret;
}
typedef void (*RedisModuleScanKeyCB)(RedisModuleKey *key, RedisModuleString *field, RedisModuleString *value, void *privdata);
typedef struct {
RedisModuleKey *key;
void* user_data;
RedisModuleScanKeyCB fn;
} ScanKeyCBData;
static void moduleScanKeyCallback(void *privdata, const dictEntry *de) {
ScanKeyCBData *data = privdata;
sds key = dictGetKey(de);
robj *o = data->key->value;
robj *field = createStringObject(key, sdslen(key));
robj *value = NULL;
if (o->type == OBJ_SET) {
value = NULL;
} else if (o->type == OBJ_HASH) {
sds val = dictGetVal(de);
value = createStringObject(val, sdslen(val));
} else if (o->type == OBJ_ZSET) {
double *val = (double*)dictGetVal(de);
value = createStringObjectFromLongDouble(*val, 0);
}
data->fn(data->key, field, value, data->user_data);
decrRefCount(field);
if (value) decrRefCount(value);
}
/* Scan api that allows a module to scan the elements in a hash, set or sorted set key
*
* Callback for scan implementation.
* void scan_callback(RedisModuleKey *key, RedisModuleString* field, RedisModuleString* value, void *privdata);
* - key - the redis key context provided to for the scan.
* - field - field name, owned by the caller and need to be retained if used
* after this function.
* - value - value string or NULL for set type, owned by the caller and need to
* be retained if used after this function.
* - privdata - the user data provided to RedisModule_ScanKey.
*
* The way it should be used:
* RedisModuleCursor *c = RedisModule_ScanCursorCreate();
* RedisModuleKey *key = RedisModule_OpenKey(...)
* while(RedisModule_ScanKey(key, c, callback, privateData));
* RedisModule_CloseKey(key);
* RedisModule_ScanCursorDestroy(c);
*
* It is also possible to use this API from another thread while the lock is acquired durring
* the actuall call to RM_Scan, and re-opening the key each time:
* RedisModuleCursor *c = RedisModule_ScanCursorCreate();
* RedisModule_ThreadSafeContextLock(ctx);
* RedisModuleKey *key = RedisModule_OpenKey(...)
* while(RedisModule_ScanKey(ctx, c, callback, privateData)){
* RedisModule_CloseKey(key);
* RedisModule_ThreadSafeContextUnlock(ctx);
* // do some background job
* RedisModule_ThreadSafeContextLock(ctx);
* RedisModuleKey *key = RedisModule_OpenKey(...)
* }
* RedisModule_CloseKey(key);
* RedisModule_ScanCursorDestroy(c);
*
* The function will return 1 if there are more elements to scan and 0 otherwise,
* possibly setting errno if the call failed.
* It is also possible to restart and existing cursor using RM_CursorRestart. */
int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleScanKeyCB fn, void *privdata) {
if (key == NULL || key->value == NULL) {
errno = EINVAL;
return 0;
}
dict *ht = NULL;
robj *o = key->value;
if (o->type == OBJ_SET) {
if (o->encoding == OBJ_ENCODING_HT)
ht = o->ptr;
} else if (o->type == OBJ_HASH) {
if (o->encoding == OBJ_ENCODING_HT)
ht = o->ptr;
} else if (o->type == OBJ_ZSET) {
if (o->encoding == OBJ_ENCODING_SKIPLIST)
ht = ((zset *)o->ptr)->dict;
} else {
errno = EINVAL;
return 0;
}
if (cursor->done) {
errno = ENOENT;
return 0;
}
int ret = 1;
if (ht) {
ScanKeyCBData data = { key, privdata, fn };
cursor->cursor = dictScan(ht, cursor->cursor, moduleScanKeyCallback, NULL, &data);
if (cursor->cursor == 0) {
cursor->done = 1;
ret = 0;
}
} else if (o->type == OBJ_SET && o->encoding == OBJ_ENCODING_INTSET) {
int pos = 0;
int64_t ll;
while(intsetGet(o->ptr,pos++,&ll)) {
robj *field = createStringObjectFromLongLong(ll);
fn(key, field, NULL, privdata);
decrRefCount(field);
}
cursor->cursor = 1;
cursor->done = 1;
ret = 0;
} else if (o->type == OBJ_HASH || o->type == OBJ_ZSET) {
unsigned char *p = ziplistIndex(o->ptr,0);
unsigned char *vstr;
unsigned int vlen;
long long vll;
while(p) {
ziplistGet(p,&vstr,&vlen,&vll);
robj *field = (vstr != NULL) ?
createStringObject((char*)vstr,vlen) :
createStringObjectFromLongLong(vll);
p = ziplistNext(o->ptr,p);
ziplistGet(p,&vstr,&vlen,&vll);
robj *value = (vstr != NULL) ?
createStringObject((char*)vstr,vlen) :
createStringObjectFromLongLong(vll);
fn(key, field, value, privdata);
p = ziplistNext(o->ptr,p);
decrRefCount(field);
decrRefCount(value);
}
cursor->cursor = 1;
cursor->done = 1;
ret = 0;
}
errno = 0;
return ret;
}
/* -------------------------------------------------------------------------- /* --------------------------------------------------------------------------
* Module fork API * Module fork API
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
...@@ -6756,38 +7093,82 @@ size_t moduleCount(void) { ...@@ -6756,38 +7093,82 @@ size_t moduleCount(void) {
return dictSize(modules); return dictSize(modules);
} }
/* Set the key LRU/LFU depending on server.maxmemory_policy. /* Set the key last access time for LRU based eviction. not relevent if the
* The lru_idle arg is idle time in seconds, and is only relevant if the * servers's maxmemory policy is LFU based. Value is idle time in milliseconds.
* eviction policy is LRU based. * returns REDISMODULE_OK if the LRU was updated, REDISMODULE_ERR otherwise. */
* The lfu_freq arg is a logarithmic counter that provides an indication of int RM_SetLRU(RedisModuleKey *key, mstime_t lru_idle) {
* the access frequencyonly (must be <= 255) and is only relevant if the
* eviction policy is LFU based.
* Either or both of them may be <0, in that case, nothing is set. */
/* return value is an indication if the lru field was updated or not. */
int RM_SetLRUOrLFU(RedisModuleKey *key, long long lfu_freq, long long lru_idle) {
if (!key->value) if (!key->value)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (objectSetLRUOrLFU(key->value, lfu_freq, lru_idle, lru_idle>=0 ? LRU_CLOCK() : 0)) if (objectSetLRUOrLFU(key->value, -1, lru_idle, lru_idle>=0 ? LRU_CLOCK() : 0, 1))
return REDISMODULE_OK; return REDISMODULE_OK;
return REDISMODULE_ERR; return REDISMODULE_ERR;
} }
/* Gets the key LRU or LFU (depending on the current eviction policy). /* Gets the key last access time.
* One will be set to the appropiate return value, and the other will be set to -1. * Value is idletime in milliseconds or -1 if the server's eviction policy is
* see RedisModule_SetLRUOrLFU for units and ranges. * LFU based.
* return value is an indication of success. */ * returns REDISMODULE_OK if when key is valid. */
int RM_GetLRUOrLFU(RedisModuleKey *key, long long *lfu_freq, long long *lru_idle) { int RM_GetLRU(RedisModuleKey *key, mstime_t *lru_idle) {
*lru_idle = *lfu_freq = -1; *lru_idle = -1;
if (!key->value) if (!key->value)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) { if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU)
return REDISMODULE_OK;
*lru_idle = estimateObjectIdleTime(key->value);
return REDISMODULE_OK;
}
/* Set the key access frequency. only relevant if the server's maxmemory policy
* is LFU based.
* The frequency is a logarithmic counter that provides an indication of
* the access frequencyonly (must be <= 255).
* returns REDISMODULE_OK if the LFU was updated, REDISMODULE_ERR otherwise. */
int RM_SetLFU(RedisModuleKey *key, long long lfu_freq) {
if (!key->value)
return REDISMODULE_ERR;
if (objectSetLRUOrLFU(key->value, lfu_freq, -1, 0, 1))
return REDISMODULE_OK;
return REDISMODULE_ERR;
}
/* Gets the key access frequency or -1 if the server's eviction policy is not
* LFU based.
* returns REDISMODULE_OK if when key is valid. */
int RM_GetLFU(RedisModuleKey *key, long long *lfu_freq) {
*lfu_freq = -1;
if (!key->value)
return REDISMODULE_ERR;
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU)
*lfu_freq = LFUDecrAndReturn(key->value); *lfu_freq = LFUDecrAndReturn(key->value);
} else {
*lru_idle = estimateObjectIdleTime(key->value)/1000;
}
return REDISMODULE_OK; return REDISMODULE_OK;
} }
/* Replace the value assigned to a module type.
*
* The key must be open for writing, have an existing value, and have a moduleType
* that matches the one specified by the caller.
*
* Unlike RM_ModuleTypeSetValue() which will free the old value, this function
* simply swaps the old value with the new value.
*
* The function returns the old value, or NULL if any of the above conditions is
* not met.
*/
void *RM_ModuleTypeReplaceValue(RedisModuleKey *key, moduleType *mt, void *new_value) {
if (!(key->mode & REDISMODULE_WRITE) || key->iter)
return NULL;
if (!key->value || key->value->type != OBJ_MODULE)
return NULL;
moduleValue *mv = key->value->ptr;
if (mv->type != mt)
return NULL;
void *old_val = mv->value;
mv->value = new_value;
return old_val;
}
/* Register all the APIs we export. Keep this function at the end of the /* Register all the APIs we export. Keep this function at the end of the
* file so that's easy to seek it to add new entries. */ * file so that's easy to seek it to add new entries. */
void moduleRegisterCoreAPI(void) { void moduleRegisterCoreAPI(void) {
...@@ -6817,6 +7198,7 @@ void moduleRegisterCoreAPI(void) { ...@@ -6817,6 +7198,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(ReplyWithNull); REGISTER_API(ReplyWithNull);
REGISTER_API(ReplyWithCallReply); REGISTER_API(ReplyWithCallReply);
REGISTER_API(ReplyWithDouble); REGISTER_API(ReplyWithDouble);
REGISTER_API(ReplyWithLongDouble);
REGISTER_API(GetSelectedDb); REGISTER_API(GetSelectedDb);
REGISTER_API(SelectDb); REGISTER_API(SelectDb);
REGISTER_API(OpenKey); REGISTER_API(OpenKey);
...@@ -6827,6 +7209,7 @@ void moduleRegisterCoreAPI(void) { ...@@ -6827,6 +7209,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(ListPop); REGISTER_API(ListPop);
REGISTER_API(StringToLongLong); REGISTER_API(StringToLongLong);
REGISTER_API(StringToDouble); REGISTER_API(StringToDouble);
REGISTER_API(StringToLongDouble);
REGISTER_API(Call); REGISTER_API(Call);
REGISTER_API(CallReplyProto); REGISTER_API(CallReplyProto);
REGISTER_API(FreeCallReply); REGISTER_API(FreeCallReply);
...@@ -6838,6 +7221,7 @@ void moduleRegisterCoreAPI(void) { ...@@ -6838,6 +7221,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(CreateStringFromCallReply); REGISTER_API(CreateStringFromCallReply);
REGISTER_API(CreateString); REGISTER_API(CreateString);
REGISTER_API(CreateStringFromLongLong); REGISTER_API(CreateStringFromLongLong);
REGISTER_API(CreateStringFromLongDouble);
REGISTER_API(CreateStringFromString); REGISTER_API(CreateStringFromString);
REGISTER_API(CreateStringPrintf); REGISTER_API(CreateStringPrintf);
REGISTER_API(FreeString); REGISTER_API(FreeString);
...@@ -6877,6 +7261,7 @@ void moduleRegisterCoreAPI(void) { ...@@ -6877,6 +7261,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(PoolAlloc); REGISTER_API(PoolAlloc);
REGISTER_API(CreateDataType); REGISTER_API(CreateDataType);
REGISTER_API(ModuleTypeSetValue); REGISTER_API(ModuleTypeSetValue);
REGISTER_API(ModuleTypeReplaceValue);
REGISTER_API(ModuleTypeGetType); REGISTER_API(ModuleTypeGetType);
REGISTER_API(ModuleTypeGetValue); REGISTER_API(ModuleTypeGetValue);
REGISTER_API(IsIOError); REGISTER_API(IsIOError);
...@@ -6896,6 +7281,8 @@ void moduleRegisterCoreAPI(void) { ...@@ -6896,6 +7281,8 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(LoadFloat); REGISTER_API(LoadFloat);
REGISTER_API(SaveLongDouble); REGISTER_API(SaveLongDouble);
REGISTER_API(LoadLongDouble); REGISTER_API(LoadLongDouble);
REGISTER_API(SaveDataTypeToString);
REGISTER_API(LoadDataTypeFromString);
REGISTER_API(EmitAOF); REGISTER_API(EmitAOF);
REGISTER_API(Log); REGISTER_API(Log);
REGISTER_API(LogIOError); REGISTER_API(LogIOError);
...@@ -6986,11 +7373,18 @@ void moduleRegisterCoreAPI(void) { ...@@ -6986,11 +7373,18 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(GetClientInfoById); REGISTER_API(GetClientInfoById);
REGISTER_API(PublishMessage); REGISTER_API(PublishMessage);
REGISTER_API(SubscribeToServerEvent); REGISTER_API(SubscribeToServerEvent);
REGISTER_API(SetLRUOrLFU); REGISTER_API(SetLRU);
REGISTER_API(GetLRUOrLFU); REGISTER_API(GetLRU);
REGISTER_API(SetLFU);
REGISTER_API(GetLFU);
REGISTER_API(BlockClientOnKeys); REGISTER_API(BlockClientOnKeys);
REGISTER_API(SignalKeyAsReady); REGISTER_API(SignalKeyAsReady);
REGISTER_API(GetBlockedClientReadyKey); REGISTER_API(GetBlockedClientReadyKey);
REGISTER_API(GetUsedMemoryRatio); REGISTER_API(GetUsedMemoryRatio);
REGISTER_API(MallocSize); REGISTER_API(MallocSize);
REGISTER_API(ScanCursorCreate);
REGISTER_API(ScanCursorDestroy);
REGISTER_API(ScanCursorRestart);
REGISTER_API(Scan);
REGISTER_API(ScanKey);
} }
...@@ -1210,7 +1210,7 @@ sds getMemoryDoctorReport(void) { ...@@ -1210,7 +1210,7 @@ sds getMemoryDoctorReport(void) {
* is MAXMEMORY_FLAG_LRU. * is MAXMEMORY_FLAG_LRU.
* Either or both of them may be <0, in that case, nothing is set. */ * Either or both of them may be <0, in that case, nothing is set. */
int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle, int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
long long lru_clock) { long long lru_clock, int lru_multiplier) {
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) { if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
if (lfu_freq >= 0) { if (lfu_freq >= 0) {
serverAssert(lfu_freq <= 255); serverAssert(lfu_freq <= 255);
...@@ -1222,7 +1222,7 @@ int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle, ...@@ -1222,7 +1222,7 @@ int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
* according to the LRU clock resolution this Redis * according to the LRU clock resolution this Redis
* instance was compiled with (normally 1000 ms, so the * instance was compiled with (normally 1000 ms, so the
* below statement will expand to lru_idle*1000/1000. */ * below statement will expand to lru_idle*1000/1000. */
lru_idle = lru_idle*1000/LRU_CLOCK_RESOLUTION; lru_idle = lru_idle*lru_multiplier/LRU_CLOCK_RESOLUTION;
long lru_abs = lru_clock - lru_idle; /* Absolute access time. */ long lru_abs = lru_clock - lru_idle; /* Absolute access time. */
/* If the LRU field underflows (since LRU it is a wrapping /* If the LRU field underflows (since LRU it is a wrapping
* clock), the best we can do is to provide a large enough LRU * clock), the best we can do is to provide a large enough LRU
......
...@@ -1673,6 +1673,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) { ...@@ -1673,6 +1673,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) {
* node, but will be our match, representing the key "f". * node, but will be our match, representing the key "f".
* *
* So in that case, we don't seek backward. */ * So in that case, we don't seek backward. */
it->data = raxGetData(it->node);
} else { } else {
if (gt && !raxIteratorNextStep(it,0)) return 0; if (gt && !raxIteratorNextStep(it,0)) return 0;
if (lt && !raxIteratorPrevStep(it,0)) return 0; if (lt && !raxIteratorPrevStep(it,0)) return 0;
...@@ -1791,7 +1792,7 @@ int raxCompare(raxIterator *iter, const char *op, unsigned char *key, size_t key ...@@ -1791,7 +1792,7 @@ int raxCompare(raxIterator *iter, const char *op, unsigned char *key, size_t key
if (eq && key_len == iter->key_len) return 1; if (eq && key_len == iter->key_len) return 1;
else if (lt) return iter->key_len < key_len; else if (lt) return iter->key_len < key_len;
else if (gt) return iter->key_len > key_len; else if (gt) return iter->key_len > key_len;
return 0; else return 0; /* Avoid warning, just 'eq' is handled before. */
} else if (cmp > 0) { } else if (cmp > 0) {
return gt ? 1 : 0; return gt ? 1 : 0;
} else /* (cmp < 0) */ { } else /* (cmp < 0) */ {
......
...@@ -2006,7 +2006,7 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) { ...@@ -2006,7 +2006,7 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
/* The DB can take some non trivial amount of time to load. Update /* The DB can take some non trivial amount of time to load. Update
* our cached time since it is used to create and update the last * our cached time since it is used to create and update the last
* interaction time with clients and for other important things. */ * interaction time with clients and for other important things. */
updateCachedTime(); updateCachedTime(0);
if (server.masterhost && server.repl_state == REPL_STATE_TRANSFER) if (server.masterhost && server.repl_state == REPL_STATE_TRANSFER)
replicationSendNewlineToMaster(); replicationSendNewlineToMaster();
loadingProgress(r->processed_bytes); loadingProgress(r->processed_bytes);
...@@ -2239,7 +2239,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) { ...@@ -2239,7 +2239,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
if (expiretime != -1) setExpire(NULL,db,key,expiretime); if (expiretime != -1) setExpire(NULL,db,key,expiretime);
/* Set usage information (for eviction). */ /* Set usage information (for eviction). */
objectSetLRUOrLFU(val,lfu_freq,lru_idle,lru_clock); objectSetLRUOrLFU(val,lfu_freq,lru_idle,lru_clock,1000);
/* Decrement the key refcount since dbAdd() will take its /* Decrement the key refcount since dbAdd() will take its
* own reference. */ * own reference. */
......
...@@ -392,6 +392,7 @@ typedef struct RedisModuleDictIter RedisModuleDictIter; ...@@ -392,6 +392,7 @@ typedef struct RedisModuleDictIter RedisModuleDictIter;
typedef struct RedisModuleCommandFilterCtx RedisModuleCommandFilterCtx; typedef struct RedisModuleCommandFilterCtx RedisModuleCommandFilterCtx;
typedef struct RedisModuleCommandFilter RedisModuleCommandFilter; typedef struct RedisModuleCommandFilter RedisModuleCommandFilter;
typedef struct RedisModuleInfoCtx RedisModuleInfoCtx; typedef struct RedisModuleInfoCtx RedisModuleInfoCtx;
typedef struct RedisModuleScanCursor RedisModuleScanCursor;
typedef int (*RedisModuleCmdFunc)(RedisModuleCtx *ctx, RedisModuleString **argv, int argc); typedef int (*RedisModuleCmdFunc)(RedisModuleCtx *ctx, RedisModuleString **argv, int argc);
typedef void (*RedisModuleDisconnectFunc)(RedisModuleCtx *ctx, RedisModuleBlockedClient *bc); typedef void (*RedisModuleDisconnectFunc)(RedisModuleCtx *ctx, RedisModuleBlockedClient *bc);
...@@ -409,6 +410,8 @@ typedef void (*RedisModuleTimerProc)(RedisModuleCtx *ctx, void *data); ...@@ -409,6 +410,8 @@ typedef void (*RedisModuleTimerProc)(RedisModuleCtx *ctx, void *data);
typedef void (*RedisModuleCommandFilterFunc) (RedisModuleCommandFilterCtx *filter); typedef void (*RedisModuleCommandFilterFunc) (RedisModuleCommandFilterCtx *filter);
typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data); typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data);
typedef void (*RedisModuleInfoFunc)(RedisModuleInfoCtx *ctx, int for_crash_report); typedef void (*RedisModuleInfoFunc)(RedisModuleInfoCtx *ctx, int for_crash_report);
typedef void (*RedisModuleScanCB)(RedisModuleCtx *ctx, RedisModuleString *keyname, RedisModuleKey *key, void *privdata);
typedef void (*RedisModuleScanKeyCB)(RedisModuleKey *key, RedisModuleString *field, RedisModuleString *value, void *privdata);
#define REDISMODULE_TYPE_METHOD_VERSION 2 #define REDISMODULE_TYPE_METHOD_VERSION 2
typedef struct RedisModuleTypeMethods { typedef struct RedisModuleTypeMethods {
...@@ -458,6 +461,7 @@ size_t REDISMODULE_API_FUNC(RedisModule_CallReplyLength)(RedisModuleCallReply *r ...@@ -458,6 +461,7 @@ size_t REDISMODULE_API_FUNC(RedisModule_CallReplyLength)(RedisModuleCallReply *r
RedisModuleCallReply *REDISMODULE_API_FUNC(RedisModule_CallReplyArrayElement)(RedisModuleCallReply *reply, size_t idx); RedisModuleCallReply *REDISMODULE_API_FUNC(RedisModule_CallReplyArrayElement)(RedisModuleCallReply *reply, size_t idx);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateString)(RedisModuleCtx *ctx, const char *ptr, size_t len); RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateString)(RedisModuleCtx *ctx, const char *ptr, size_t len);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromLongLong)(RedisModuleCtx *ctx, long long ll); RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromLongLong)(RedisModuleCtx *ctx, long long ll);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromLongDouble)(RedisModuleCtx *ctx, long double ld, int humanfriendly);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromString)(RedisModuleCtx *ctx, const RedisModuleString *str); RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromString)(RedisModuleCtx *ctx, const RedisModuleString *str);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringPrintf)(RedisModuleCtx *ctx, const char *fmt, ...); RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringPrintf)(RedisModuleCtx *ctx, const char *fmt, ...);
void REDISMODULE_API_FUNC(RedisModule_FreeString)(RedisModuleCtx *ctx, RedisModuleString *str); void REDISMODULE_API_FUNC(RedisModule_FreeString)(RedisModuleCtx *ctx, RedisModuleString *str);
...@@ -475,9 +479,11 @@ int REDISMODULE_API_FUNC(RedisModule_ReplyWithEmptyString)(RedisModuleCtx *ctx); ...@@ -475,9 +479,11 @@ int REDISMODULE_API_FUNC(RedisModule_ReplyWithEmptyString)(RedisModuleCtx *ctx);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithVerbatimString)(RedisModuleCtx *ctx, const char *buf, size_t len); int REDISMODULE_API_FUNC(RedisModule_ReplyWithVerbatimString)(RedisModuleCtx *ctx, const char *buf, size_t len);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithNull)(RedisModuleCtx *ctx); int REDISMODULE_API_FUNC(RedisModule_ReplyWithNull)(RedisModuleCtx *ctx);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithDouble)(RedisModuleCtx *ctx, double d); int REDISMODULE_API_FUNC(RedisModule_ReplyWithDouble)(RedisModuleCtx *ctx, double d);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithLongDouble)(RedisModuleCtx *ctx, long double d);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, RedisModuleCallReply *reply); int REDISMODULE_API_FUNC(RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, RedisModuleCallReply *reply);
int REDISMODULE_API_FUNC(RedisModule_StringToLongLong)(const RedisModuleString *str, long long *ll); int REDISMODULE_API_FUNC(RedisModule_StringToLongLong)(const RedisModuleString *str, long long *ll);
int REDISMODULE_API_FUNC(RedisModule_StringToDouble)(const RedisModuleString *str, double *d); int REDISMODULE_API_FUNC(RedisModule_StringToDouble)(const RedisModuleString *str, double *d);
int REDISMODULE_API_FUNC(RedisModule_StringToLongDouble)(const RedisModuleString *str, long double *d);
void REDISMODULE_API_FUNC(RedisModule_AutoMemory)(RedisModuleCtx *ctx); void REDISMODULE_API_FUNC(RedisModule_AutoMemory)(RedisModuleCtx *ctx);
int REDISMODULE_API_FUNC(RedisModule_Replicate)(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...); int REDISMODULE_API_FUNC(RedisModule_Replicate)(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...);
int REDISMODULE_API_FUNC(RedisModule_ReplicateVerbatim)(RedisModuleCtx *ctx); int REDISMODULE_API_FUNC(RedisModule_ReplicateVerbatim)(RedisModuleCtx *ctx);
...@@ -517,6 +523,7 @@ int REDISMODULE_API_FUNC(RedisModule_GetContextFlags)(RedisModuleCtx *ctx); ...@@ -517,6 +523,7 @@ int REDISMODULE_API_FUNC(RedisModule_GetContextFlags)(RedisModuleCtx *ctx);
void *REDISMODULE_API_FUNC(RedisModule_PoolAlloc)(RedisModuleCtx *ctx, size_t bytes); void *REDISMODULE_API_FUNC(RedisModule_PoolAlloc)(RedisModuleCtx *ctx, size_t bytes);
RedisModuleType *REDISMODULE_API_FUNC(RedisModule_CreateDataType)(RedisModuleCtx *ctx, const char *name, int encver, RedisModuleTypeMethods *typemethods); RedisModuleType *REDISMODULE_API_FUNC(RedisModule_CreateDataType)(RedisModuleCtx *ctx, const char *name, int encver, RedisModuleTypeMethods *typemethods);
int REDISMODULE_API_FUNC(RedisModule_ModuleTypeSetValue)(RedisModuleKey *key, RedisModuleType *mt, void *value); int REDISMODULE_API_FUNC(RedisModule_ModuleTypeSetValue)(RedisModuleKey *key, RedisModuleType *mt, void *value);
void *REDISMODULE_API_FUNC(RedisModule_ModuleTypeReplaceValue)(RedisModuleKey *key, RedisModuleType *mt, void *new_value);
RedisModuleType *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetType)(RedisModuleKey *key); RedisModuleType *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetType)(RedisModuleKey *key);
void *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetValue)(RedisModuleKey *key); void *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetValue)(RedisModuleKey *key);
int REDISMODULE_API_FUNC(RedisModule_IsIOError)(RedisModuleIO *io); int REDISMODULE_API_FUNC(RedisModule_IsIOError)(RedisModuleIO *io);
...@@ -537,6 +544,8 @@ void REDISMODULE_API_FUNC(RedisModule_SaveFloat)(RedisModuleIO *io, float value) ...@@ -537,6 +544,8 @@ void REDISMODULE_API_FUNC(RedisModule_SaveFloat)(RedisModuleIO *io, float value)
float REDISMODULE_API_FUNC(RedisModule_LoadFloat)(RedisModuleIO *io); float REDISMODULE_API_FUNC(RedisModule_LoadFloat)(RedisModuleIO *io);
void REDISMODULE_API_FUNC(RedisModule_SaveLongDouble)(RedisModuleIO *io, long double value); void REDISMODULE_API_FUNC(RedisModule_SaveLongDouble)(RedisModuleIO *io, long double value);
long double REDISMODULE_API_FUNC(RedisModule_LoadLongDouble)(RedisModuleIO *io); long double REDISMODULE_API_FUNC(RedisModule_LoadLongDouble)(RedisModuleIO *io);
void *REDISMODULE_API_FUNC(RedisModule_LoadDataTypeFromString)(const RedisModuleString *str, const RedisModuleType *mt);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_SaveDataTypeToString)(RedisModuleCtx *ctx, void *data, const RedisModuleType *mt);
void REDISMODULE_API_FUNC(RedisModule_Log)(RedisModuleCtx *ctx, const char *level, const char *fmt, ...); void REDISMODULE_API_FUNC(RedisModule_Log)(RedisModuleCtx *ctx, const char *level, const char *fmt, ...);
void REDISMODULE_API_FUNC(RedisModule_LogIOError)(RedisModuleIO *io, const char *levelstr, const char *fmt, ...); void REDISMODULE_API_FUNC(RedisModule_LogIOError)(RedisModuleIO *io, const char *levelstr, const char *fmt, ...);
void REDISMODULE_API_FUNC(RedisModule__Assert)(const char *estr, const char *file, int line); void REDISMODULE_API_FUNC(RedisModule__Assert)(const char *estr, const char *file, int line);
...@@ -583,11 +592,18 @@ int REDISMODULE_API_FUNC(RedisModule_InfoAddFieldDouble)(RedisModuleInfoCtx *ctx ...@@ -583,11 +592,18 @@ int REDISMODULE_API_FUNC(RedisModule_InfoAddFieldDouble)(RedisModuleInfoCtx *ctx
int REDISMODULE_API_FUNC(RedisModule_InfoAddFieldLongLong)(RedisModuleInfoCtx *ctx, char *field, long long value); int REDISMODULE_API_FUNC(RedisModule_InfoAddFieldLongLong)(RedisModuleInfoCtx *ctx, char *field, long long value);
int REDISMODULE_API_FUNC(RedisModule_InfoAddFieldULongLong)(RedisModuleInfoCtx *ctx, char *field, unsigned long long value); int REDISMODULE_API_FUNC(RedisModule_InfoAddFieldULongLong)(RedisModuleInfoCtx *ctx, char *field, unsigned long long value);
int REDISMODULE_API_FUNC(RedisModule_SubscribeToServerEvent)(RedisModuleCtx *ctx, RedisModuleEvent event, RedisModuleEventCallback callback); int REDISMODULE_API_FUNC(RedisModule_SubscribeToServerEvent)(RedisModuleCtx *ctx, RedisModuleEvent event, RedisModuleEventCallback callback);
int REDISMODULE_API_FUNC(RedisModule_SetLRUOrLFU)(RedisModuleKey *key, long long lfu_freq, long long lru_idle); int REDISMODULE_API_FUNC(RedisModule_SetLRU)(RedisModuleKey *key, mstime_t lru_idle);
int REDISMODULE_API_FUNC(RedisModule_GetLRUOrLFU)(RedisModuleKey *key, long long *lfu_freq, long long *lru_idle); int REDISMODULE_API_FUNC(RedisModule_GetLRU)(RedisModuleKey *key, mstime_t *lru_idle);
int REDISMODULE_API_FUNC(RedisModule_SetLFU)(RedisModuleKey *key, long long lfu_freq);
int REDISMODULE_API_FUNC(RedisModule_GetLFU)(RedisModuleKey *key, long long *lfu_freq);
RedisModuleBlockedClient *REDISMODULE_API_FUNC(RedisModule_BlockClientOnKeys)(RedisModuleCtx *ctx, RedisModuleCmdFunc reply_callback, RedisModuleCmdFunc timeout_callback, void (*free_privdata)(RedisModuleCtx*,void*), long long timeout_ms, RedisModuleString **keys, int numkeys, void *privdata); RedisModuleBlockedClient *REDISMODULE_API_FUNC(RedisModule_BlockClientOnKeys)(RedisModuleCtx *ctx, RedisModuleCmdFunc reply_callback, RedisModuleCmdFunc timeout_callback, void (*free_privdata)(RedisModuleCtx*,void*), long long timeout_ms, RedisModuleString **keys, int numkeys, void *privdata);
void REDISMODULE_API_FUNC(RedisModule_SignalKeyAsReady)(RedisModuleCtx *ctx, RedisModuleString *key); void REDISMODULE_API_FUNC(RedisModule_SignalKeyAsReady)(RedisModuleCtx *ctx, RedisModuleString *key);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_GetBlockedClientReadyKey)(RedisModuleCtx *ctx); RedisModuleString *REDISMODULE_API_FUNC(RedisModule_GetBlockedClientReadyKey)(RedisModuleCtx *ctx);
RedisModuleScanCursor *REDISMODULE_API_FUNC(RedisModule_ScanCursorCreate)();
void REDISMODULE_API_FUNC(RedisModule_ScanCursorRestart)(RedisModuleScanCursor *cursor);
void REDISMODULE_API_FUNC(RedisModule_ScanCursorDestroy)(RedisModuleScanCursor *cursor);
int REDISMODULE_API_FUNC(RedisModule_Scan)(RedisModuleCtx *ctx, RedisModuleScanCursor *cursor, RedisModuleScanCB fn, void *privdata);
int REDISMODULE_API_FUNC(RedisModule_ScanKey)(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleScanKeyCB fn, void *privdata);
/* Experimental APIs */ /* Experimental APIs */
#ifdef REDISMODULE_EXPERIMENTAL_API #ifdef REDISMODULE_EXPERIMENTAL_API
...@@ -668,6 +684,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -668,6 +684,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(ReplyWithNull); REDISMODULE_GET_API(ReplyWithNull);
REDISMODULE_GET_API(ReplyWithCallReply); REDISMODULE_GET_API(ReplyWithCallReply);
REDISMODULE_GET_API(ReplyWithDouble); REDISMODULE_GET_API(ReplyWithDouble);
REDISMODULE_GET_API(ReplyWithLongDouble);
REDISMODULE_GET_API(GetSelectedDb); REDISMODULE_GET_API(GetSelectedDb);
REDISMODULE_GET_API(SelectDb); REDISMODULE_GET_API(SelectDb);
REDISMODULE_GET_API(OpenKey); REDISMODULE_GET_API(OpenKey);
...@@ -678,6 +695,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -678,6 +695,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(ListPop); REDISMODULE_GET_API(ListPop);
REDISMODULE_GET_API(StringToLongLong); REDISMODULE_GET_API(StringToLongLong);
REDISMODULE_GET_API(StringToDouble); REDISMODULE_GET_API(StringToDouble);
REDISMODULE_GET_API(StringToLongDouble);
REDISMODULE_GET_API(Call); REDISMODULE_GET_API(Call);
REDISMODULE_GET_API(CallReplyProto); REDISMODULE_GET_API(CallReplyProto);
REDISMODULE_GET_API(FreeCallReply); REDISMODULE_GET_API(FreeCallReply);
...@@ -689,6 +707,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -689,6 +707,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(CreateStringFromCallReply); REDISMODULE_GET_API(CreateStringFromCallReply);
REDISMODULE_GET_API(CreateString); REDISMODULE_GET_API(CreateString);
REDISMODULE_GET_API(CreateStringFromLongLong); REDISMODULE_GET_API(CreateStringFromLongLong);
REDISMODULE_GET_API(CreateStringFromLongDouble);
REDISMODULE_GET_API(CreateStringFromString); REDISMODULE_GET_API(CreateStringFromString);
REDISMODULE_GET_API(CreateStringPrintf); REDISMODULE_GET_API(CreateStringPrintf);
REDISMODULE_GET_API(FreeString); REDISMODULE_GET_API(FreeString);
...@@ -728,6 +747,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -728,6 +747,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(PoolAlloc); REDISMODULE_GET_API(PoolAlloc);
REDISMODULE_GET_API(CreateDataType); REDISMODULE_GET_API(CreateDataType);
REDISMODULE_GET_API(ModuleTypeSetValue); REDISMODULE_GET_API(ModuleTypeSetValue);
REDISMODULE_GET_API(ModuleTypeReplaceValue);
REDISMODULE_GET_API(ModuleTypeGetType); REDISMODULE_GET_API(ModuleTypeGetType);
REDISMODULE_GET_API(ModuleTypeGetValue); REDISMODULE_GET_API(ModuleTypeGetValue);
REDISMODULE_GET_API(IsIOError); REDISMODULE_GET_API(IsIOError);
...@@ -747,6 +767,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -747,6 +767,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(LoadFloat); REDISMODULE_GET_API(LoadFloat);
REDISMODULE_GET_API(SaveLongDouble); REDISMODULE_GET_API(SaveLongDouble);
REDISMODULE_GET_API(LoadLongDouble); REDISMODULE_GET_API(LoadLongDouble);
REDISMODULE_GET_API(SaveDataTypeToString);
REDISMODULE_GET_API(LoadDataTypeFromString);
REDISMODULE_GET_API(EmitAOF); REDISMODULE_GET_API(EmitAOF);
REDISMODULE_GET_API(Log); REDISMODULE_GET_API(Log);
REDISMODULE_GET_API(LogIOError); REDISMODULE_GET_API(LogIOError);
...@@ -796,11 +818,18 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -796,11 +818,18 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(GetClientInfoById); REDISMODULE_GET_API(GetClientInfoById);
REDISMODULE_GET_API(PublishMessage); REDISMODULE_GET_API(PublishMessage);
REDISMODULE_GET_API(SubscribeToServerEvent); REDISMODULE_GET_API(SubscribeToServerEvent);
REDISMODULE_GET_API(SetLRUOrLFU); REDISMODULE_GET_API(SetLRU);
REDISMODULE_GET_API(GetLRUOrLFU); REDISMODULE_GET_API(GetLRU);
REDISMODULE_GET_API(SetLFU);
REDISMODULE_GET_API(GetLFU);
REDISMODULE_GET_API(BlockClientOnKeys); REDISMODULE_GET_API(BlockClientOnKeys);
REDISMODULE_GET_API(SignalKeyAsReady); REDISMODULE_GET_API(SignalKeyAsReady);
REDISMODULE_GET_API(GetBlockedClientReadyKey); REDISMODULE_GET_API(GetBlockedClientReadyKey);
REDISMODULE_GET_API(ScanCursorCreate);
REDISMODULE_GET_API(ScanCursorRestart);
REDISMODULE_GET_API(ScanCursorDestroy);
REDISMODULE_GET_API(Scan);
REDISMODULE_GET_API(ScanKey);
#ifdef REDISMODULE_EXPERIMENTAL_API #ifdef REDISMODULE_EXPERIMENTAL_API
REDISMODULE_GET_API(GetThreadSafeContext); REDISMODULE_GET_API(GetThreadSafeContext);
......
...@@ -3993,11 +3993,14 @@ int sentinelSendSlaveOf(sentinelRedisInstance *ri, char *host, int port) { ...@@ -3993,11 +3993,14 @@ int sentinelSendSlaveOf(sentinelRedisInstance *ri, char *host, int port) {
* an issue because CLIENT is variadic command, so Redis will not * an issue because CLIENT is variadic command, so Redis will not
* recognized as a syntax error, and the transaction will not fail (but * recognized as a syntax error, and the transaction will not fail (but
* only the unsupported command will fail). */ * only the unsupported command will fail). */
retval = redisAsyncCommand(ri->link->cc, for (int type = 0; type < 2; type++) {
sentinelDiscardReplyCallback, ri, "%s KILL TYPE normal", retval = redisAsyncCommand(ri->link->cc,
sentinelInstanceMapCommand(ri,"CLIENT")); sentinelDiscardReplyCallback, ri, "%s KILL TYPE %s",
if (retval == C_ERR) return retval; sentinelInstanceMapCommand(ri,"CLIENT"),
ri->link->pending_commands++; type == 0 ? "normal" : "pubsub");
if (retval == C_ERR) return retval;
ri->link->pending_commands++;
}
retval = redisAsyncCommand(ri->link->cc, retval = redisAsyncCommand(ri->link->cc,
sentinelDiscardReplyCallback, ri, "%s", sentinelDiscardReplyCallback, ri, "%s",
......
...@@ -1691,8 +1691,7 @@ void databasesCron(void) { ...@@ -1691,8 +1691,7 @@ void databasesCron(void) {
} }
/* Defrag keys gradually. */ /* Defrag keys gradually. */
if (server.active_defrag_enabled) activeDefragCycle();
activeDefragCycle();
/* Perform hash tables rehashing if needed, but only if there are no /* Perform hash tables rehashing if needed, but only if there are no
* other processes saving the DB on disk. Otherwise rehashing is bad * other processes saving the DB on disk. Otherwise rehashing is bad
...@@ -1736,20 +1735,29 @@ void databasesCron(void) { ...@@ -1736,20 +1735,29 @@ void databasesCron(void) {
/* We take a cached value of the unix time in the global state because with /* We take a cached value of the unix time in the global state because with
* virtual memory and aging there is to store the current time in objects at * virtual memory and aging there is to store the current time in objects at
* every object access, and accuracy is not needed. To access a global var is * every object access, and accuracy is not needed. To access a global var is
* a lot faster than calling time(NULL) */ * a lot faster than calling time(NULL).
void updateCachedTime(void) { *
server.unixtime = time(NULL); * This function should be fast because it is called at every command execution
server.mstime = mstime(); * in call(), so it is possible to decide if to update the daylight saving
* info or not using the 'update_daylight_info' argument. Normally we update
* such info only when calling this function from serverCron() but not when
* calling it from call(). */
void updateCachedTime(int update_daylight_info) {
server.ustime = ustime();
server.mstime = server.ustime / 1000;
server.unixtime = server.mstime / 1000;
/* To get information about daylight saving time, we need to call /* To get information about daylight saving time, we need to call
* localtime_r and cache the result. However calling localtime_r in this * localtime_r and cache the result. However calling localtime_r in this
* context is safe since we will never fork() while here, in the main * context is safe since we will never fork() while here, in the main
* thread. The logging function will call a thread safe version of * thread. The logging function will call a thread safe version of
* localtime that has no locks. */ * localtime that has no locks. */
struct tm tm; if (update_daylight_info) {
time_t ut = server.unixtime; struct tm tm;
localtime_r(&ut,&tm); time_t ut = server.unixtime;
server.daylight_active = tm.tm_isdst; localtime_r(&ut,&tm);
server.daylight_active = tm.tm_isdst;
}
} }
void checkChildrenDone(void) { void checkChildrenDone(void) {
...@@ -1838,7 +1846,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { ...@@ -1838,7 +1846,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
if (server.watchdog_period) watchdogScheduleSignal(server.watchdog_period); if (server.watchdog_period) watchdogScheduleSignal(server.watchdog_period);
/* Update the time cache. */ /* Update the time cache. */
updateCachedTime(); updateCachedTime(1);
server.hz = server.config_hz; server.hz = server.config_hz;
/* Adapt the server.hz value to the number of configured clients. If we have /* Adapt the server.hz value to the number of configured clients. If we have
...@@ -2258,7 +2266,7 @@ void createSharedObjects(void) { ...@@ -2258,7 +2266,7 @@ void createSharedObjects(void) {
void initServerConfig(void) { void initServerConfig(void) {
int j; int j;
updateCachedTime(); updateCachedTime(1);
getRandomHexChars(server.runid,CONFIG_RUN_ID_SIZE); getRandomHexChars(server.runid,CONFIG_RUN_ID_SIZE);
server.runid[CONFIG_RUN_ID_SIZE] = '\0'; server.runid[CONFIG_RUN_ID_SIZE] = '\0';
changeReplicationId(); changeReplicationId();
...@@ -2285,6 +2293,7 @@ void initServerConfig(void) { ...@@ -2285,6 +2293,7 @@ void initServerConfig(void) {
server.maxidletime = CONFIG_DEFAULT_CLIENT_TIMEOUT; server.maxidletime = CONFIG_DEFAULT_CLIENT_TIMEOUT;
server.tcpkeepalive = CONFIG_DEFAULT_TCP_KEEPALIVE; server.tcpkeepalive = CONFIG_DEFAULT_TCP_KEEPALIVE;
server.active_expire_enabled = 1; server.active_expire_enabled = 1;
server.active_expire_effort = CONFIG_DEFAULT_ACTIVE_EXPIRE_EFFORT;
server.jemalloc_bg_thread = 1; server.jemalloc_bg_thread = 1;
server.active_defrag_enabled = CONFIG_DEFAULT_ACTIVE_DEFRAG; server.active_defrag_enabled = CONFIG_DEFAULT_ACTIVE_DEFRAG;
server.active_defrag_ignore_bytes = CONFIG_DEFAULT_DEFRAG_IGNORE_BYTES; server.active_defrag_ignore_bytes = CONFIG_DEFAULT_DEFRAG_IGNORE_BYTES;
...@@ -2736,6 +2745,7 @@ void resetServerStats(void) { ...@@ -2736,6 +2745,7 @@ void resetServerStats(void) {
server.stat_expiredkeys = 0; server.stat_expiredkeys = 0;
server.stat_expired_stale_perc = 0; server.stat_expired_stale_perc = 0;
server.stat_expired_time_cap_reached_count = 0; server.stat_expired_time_cap_reached_count = 0;
server.stat_expire_cycle_time_used = 0;
server.stat_evictedkeys = 0; server.stat_evictedkeys = 0;
server.stat_keyspace_misses = 0; server.stat_keyspace_misses = 0;
server.stat_keyspace_hits = 0; server.stat_keyspace_hits = 0;
...@@ -2777,6 +2787,7 @@ void initServer(void) { ...@@ -2777,6 +2787,7 @@ void initServer(void) {
server.hz = server.config_hz; server.hz = server.config_hz;
server.pid = getpid(); server.pid = getpid();
server.current_client = NULL; server.current_client = NULL;
server.fixed_time_expire = 0;
server.clients = listCreate(); server.clients = listCreate();
server.clients_index = raxNew(); server.clients_index = raxNew();
server.clients_to_close = listCreate(); server.clients_to_close = listCreate();
...@@ -2838,12 +2849,14 @@ void initServer(void) { ...@@ -2838,12 +2849,14 @@ void initServer(void) {
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
server.db[j].dict = dictCreate(&dbDictType,NULL); server.db[j].dict = dictCreate(&dbDictType,NULL);
server.db[j].expires = dictCreate(&keyptrDictType,NULL); server.db[j].expires = dictCreate(&keyptrDictType,NULL);
server.db[j].expires_cursor = 0;
server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL); server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL); server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
server.db[j].watched_keys = dictCreate(&keylistDictType,NULL); server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
server.db[j].id = j; server.db[j].id = j;
server.db[j].avg_ttl = 0; server.db[j].avg_ttl = 0;
server.db[j].defrag_later = listCreate(); server.db[j].defrag_later = listCreate();
listSetFreeMethod(server.db[j].defrag_later,(void (*)(void*))sdsfree);
} }
evictionPoolAlloc(); /* Initialize the LRU keys pool. */ evictionPoolAlloc(); /* Initialize the LRU keys pool. */
server.pubsub_channels = dictCreate(&keylistDictType,NULL); server.pubsub_channels = dictCreate(&keylistDictType,NULL);
...@@ -3244,10 +3257,13 @@ void preventCommandReplication(client *c) { ...@@ -3244,10 +3257,13 @@ void preventCommandReplication(client *c) {
* *
*/ */
void call(client *c, int flags) { void call(client *c, int flags) {
long long dirty, start, duration; long long dirty;
ustime_t start, duration;
int client_old_flags = c->flags; int client_old_flags = c->flags;
struct redisCommand *real_cmd = c->cmd; struct redisCommand *real_cmd = c->cmd;
server.fixed_time_expire++;
/* Sent the command to clients in MONITOR mode, only if the commands are /* Sent the command to clients in MONITOR mode, only if the commands are
* not generated from reading an AOF. */ * not generated from reading an AOF. */
if (listLength(server.monitors) && if (listLength(server.monitors) &&
...@@ -3265,7 +3281,8 @@ void call(client *c, int flags) { ...@@ -3265,7 +3281,8 @@ void call(client *c, int flags) {
/* Call the command. */ /* Call the command. */
dirty = server.dirty; dirty = server.dirty;
start = ustime(); updateCachedTime(0);
start = server.ustime;
c->cmd->proc(c); c->cmd->proc(c);
duration = ustime()-start; duration = ustime()-start;
dirty = server.dirty-dirty; dirty = server.dirty-dirty;
...@@ -3372,6 +3389,7 @@ void call(client *c, int flags) { ...@@ -3372,6 +3389,7 @@ void call(client *c, int flags) {
trackingRememberKeys(caller); trackingRememberKeys(caller);
} }
server.fixed_time_expire--;
server.stat_numcommands++; server.stat_numcommands++;
} }
...@@ -4253,6 +4271,7 @@ sds genRedisInfoString(char *section) { ...@@ -4253,6 +4271,7 @@ sds genRedisInfoString(char *section) {
"expired_keys:%lld\r\n" "expired_keys:%lld\r\n"
"expired_stale_perc:%.2f\r\n" "expired_stale_perc:%.2f\r\n"
"expired_time_cap_reached_count:%lld\r\n" "expired_time_cap_reached_count:%lld\r\n"
"expire_cycle_cpu_milliseconds:%lld\r\n"
"evicted_keys:%lld\r\n" "evicted_keys:%lld\r\n"
"keyspace_hits:%lld\r\n" "keyspace_hits:%lld\r\n"
"keyspace_misses:%lld\r\n" "keyspace_misses:%lld\r\n"
...@@ -4280,6 +4299,7 @@ sds genRedisInfoString(char *section) { ...@@ -4280,6 +4299,7 @@ sds genRedisInfoString(char *section) {
server.stat_expiredkeys, server.stat_expiredkeys,
server.stat_expired_stale_perc*100, server.stat_expired_stale_perc*100,
server.stat_expired_time_cap_reached_count, server.stat_expired_time_cap_reached_count,
server.stat_expire_cycle_time_used/1000,
server.stat_evictedkeys, server.stat_evictedkeys,
server.stat_keyspace_hits, server.stat_keyspace_hits,
server.stat_keyspace_misses, server.stat_keyspace_misses,
......
...@@ -50,6 +50,7 @@ ...@@ -50,6 +50,7 @@
#include <signal.h> #include <signal.h>
typedef long long mstime_t; /* millisecond time type. */ typedef long long mstime_t; /* millisecond time type. */
typedef long long ustime_t; /* microsecond time type. */
#include "ae.h" /* Event driven programming library */ #include "ae.h" /* Event driven programming library */
#include "sds.h" /* Dynamic safe strings */ #include "sds.h" /* Dynamic safe strings */
...@@ -173,15 +174,13 @@ typedef long long mstime_t; /* millisecond time type. */ ...@@ -173,15 +174,13 @@ typedef long long mstime_t; /* millisecond time type. */
#define CONFIG_DEFAULT_DEFRAG_THRESHOLD_LOWER 10 /* don't defrag when fragmentation is below 10% */ #define CONFIG_DEFAULT_DEFRAG_THRESHOLD_LOWER 10 /* don't defrag when fragmentation is below 10% */
#define CONFIG_DEFAULT_DEFRAG_THRESHOLD_UPPER 100 /* maximum defrag force at 100% fragmentation */ #define CONFIG_DEFAULT_DEFRAG_THRESHOLD_UPPER 100 /* maximum defrag force at 100% fragmentation */
#define CONFIG_DEFAULT_DEFRAG_IGNORE_BYTES (100<<20) /* don't defrag if frag overhead is below 100mb */ #define CONFIG_DEFAULT_DEFRAG_IGNORE_BYTES (100<<20) /* don't defrag if frag overhead is below 100mb */
#define CONFIG_DEFAULT_DEFRAG_CYCLE_MIN 5 /* 5% CPU min (at lower threshold) */ #define CONFIG_DEFAULT_DEFRAG_CYCLE_MIN 1 /* 1% CPU min (at lower threshold) */
#define CONFIG_DEFAULT_DEFRAG_CYCLE_MAX 75 /* 75% CPU max (at upper threshold) */ #define CONFIG_DEFAULT_DEFRAG_CYCLE_MAX 25 /* 25% CPU max (at upper threshold) */
#define CONFIG_DEFAULT_DEFRAG_MAX_SCAN_FIELDS 1000 /* keys with more than 1000 fields will be processed separately */ #define CONFIG_DEFAULT_DEFRAG_MAX_SCAN_FIELDS 1000 /* keys with more than 1000 fields will be processed separately */
#define CONFIG_DEFAULT_PROTO_MAX_BULK_LEN (512ll*1024*1024) /* Bulk request max size */ #define CONFIG_DEFAULT_PROTO_MAX_BULK_LEN (512ll*1024*1024) /* Bulk request max size */
#define CONFIG_DEFAULT_TRACKING_TABLE_MAX_FILL 10 /* 10% tracking table max fill. */ #define CONFIG_DEFAULT_TRACKING_TABLE_MAX_FILL 10 /* 10% tracking table max fill. */
#define CONFIG_DEFAULT_ACTIVE_EXPIRE_EFFORT 1 /* From 1 to 10. */
#define ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP 20 /* Loopkups per loop. */
#define ACTIVE_EXPIRE_CYCLE_FAST_DURATION 1000 /* Microseconds */
#define ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 25 /* CPU max % for keys collection */
#define ACTIVE_EXPIRE_CYCLE_SLOW 0 #define ACTIVE_EXPIRE_CYCLE_SLOW 0
#define ACTIVE_EXPIRE_CYCLE_FAST 1 #define ACTIVE_EXPIRE_CYCLE_FAST 1
...@@ -720,6 +719,7 @@ typedef struct redisDb { ...@@ -720,6 +719,7 @@ typedef struct redisDb {
dict *watched_keys; /* WATCHED keys for MULTI/EXEC CAS */ dict *watched_keys; /* WATCHED keys for MULTI/EXEC CAS */
int id; /* Database ID */ int id; /* Database ID */
long long avg_ttl; /* Average TTL, just for stats */ long long avg_ttl; /* Average TTL, just for stats */
unsigned long expires_cursor; /* Cursor of the active expire cycle. */
list *defrag_later; /* List of key names to attempt to defrag one by one, gradually. */ list *defrag_later; /* List of key names to attempt to defrag one by one, gradually. */
} redisDb; } redisDb;
...@@ -1133,7 +1133,8 @@ struct redisServer { ...@@ -1133,7 +1133,8 @@ struct redisServer {
list *clients_pending_write; /* There is to write or install handler. */ list *clients_pending_write; /* There is to write or install handler. */
list *clients_pending_read; /* Client has pending read socket buffers. */ list *clients_pending_read; /* Client has pending read socket buffers. */
list *slaves, *monitors; /* List of slaves and MONITORs */ list *slaves, *monitors; /* List of slaves and MONITORs */
client *current_client; /* Current client, only used on crash report */ client *current_client; /* Current client executing the command. */
long fixed_time_expire; /* If > 0, expire keys against server.mstime. */
rax *clients_index; /* Active clients dictionary by client ID. */ rax *clients_index; /* Active clients dictionary by client ID. */
int clients_paused; /* True if clients are currently paused */ int clients_paused; /* True if clients are currently paused */
mstime_t clients_pause_end_time; /* Time when we undo clients_paused */ mstime_t clients_pause_end_time; /* Time when we undo clients_paused */
...@@ -1165,6 +1166,7 @@ struct redisServer { ...@@ -1165,6 +1166,7 @@ struct redisServer {
long long stat_expiredkeys; /* Number of expired keys */ long long stat_expiredkeys; /* Number of expired keys */
double stat_expired_stale_perc; /* Percentage of keys probably expired */ double stat_expired_stale_perc; /* Percentage of keys probably expired */
long long stat_expired_time_cap_reached_count; /* Early expire cylce stops.*/ long long stat_expired_time_cap_reached_count; /* Early expire cylce stops.*/
long long stat_expire_cycle_time_used; /* Cumulative microseconds used. */
long long stat_evictedkeys; /* Number of evicted keys (maxmemory) */ long long stat_evictedkeys; /* Number of evicted keys (maxmemory) */
long long stat_keyspace_hits; /* Number of successful lookups of keys */ long long stat_keyspace_hits; /* Number of successful lookups of keys */
long long stat_keyspace_misses; /* Number of failed lookups of keys */ long long stat_keyspace_misses; /* Number of failed lookups of keys */
...@@ -1203,6 +1205,7 @@ struct redisServer { ...@@ -1203,6 +1205,7 @@ struct redisServer {
int maxidletime; /* Client timeout in seconds */ int maxidletime; /* Client timeout in seconds */
int tcpkeepalive; /* Set SO_KEEPALIVE if non-zero. */ int tcpkeepalive; /* Set SO_KEEPALIVE if non-zero. */
int active_expire_enabled; /* Can be disabled for testing purposes. */ int active_expire_enabled; /* Can be disabled for testing purposes. */
int active_expire_effort; /* From 1 (default) to 10, active effort. */
int active_defrag_enabled; int active_defrag_enabled;
int jemalloc_bg_thread; /* Enable jemalloc background thread */ int jemalloc_bg_thread; /* Enable jemalloc background thread */
size_t active_defrag_ignore_bytes; /* minimum amount of fragmentation waste to start active defrag */ size_t active_defrag_ignore_bytes; /* minimum amount of fragmentation waste to start active defrag */
...@@ -1400,7 +1403,8 @@ struct redisServer { ...@@ -1400,7 +1403,8 @@ struct redisServer {
_Atomic time_t unixtime; /* Unix time sampled every cron cycle. */ _Atomic time_t unixtime; /* Unix time sampled every cron cycle. */
time_t timezone; /* Cached timezone. As set by tzset(). */ time_t timezone; /* Cached timezone. As set by tzset(). */
int daylight_active; /* Currently in daylight saving time. */ int daylight_active; /* Currently in daylight saving time. */
long long mstime; /* 'unixtime' with milliseconds resolution. */ mstime_t mstime; /* 'unixtime' in milliseconds. */
ustime_t ustime; /* 'unixtime' in microseconds. */
/* Pubsub */ /* Pubsub */
dict *pubsub_channels; /* Map channels to list of subscribed clients */ dict *pubsub_channels; /* Map channels to list of subscribed clients */
list *pubsub_patterns; /* A list of pubsub_patterns */ list *pubsub_patterns; /* A list of pubsub_patterns */
...@@ -1999,7 +2003,7 @@ void populateCommandTable(void); ...@@ -1999,7 +2003,7 @@ void populateCommandTable(void);
void resetCommandTableStats(void); void resetCommandTableStats(void);
void adjustOpenFilesLimit(void); void adjustOpenFilesLimit(void);
void closeListeningSockets(int unlink_unix_socket); void closeListeningSockets(int unlink_unix_socket);
void updateCachedTime(void); void updateCachedTime(int update_daylight_info);
void resetServerStats(void); void resetServerStats(void);
void activeDefragCycle(void); void activeDefragCycle(void);
unsigned int getLRUClock(void); unsigned int getLRUClock(void);
...@@ -2089,7 +2093,7 @@ robj *lookupKeyWriteWithFlags(redisDb *db, robj *key, int flags); ...@@ -2089,7 +2093,7 @@ robj *lookupKeyWriteWithFlags(redisDb *db, robj *key, int flags);
robj *objectCommandLookup(client *c, robj *key); robj *objectCommandLookup(client *c, robj *key);
robj *objectCommandLookupOrReply(client *c, robj *key, robj *reply); robj *objectCommandLookupOrReply(client *c, robj *key, robj *reply);
int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle, int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
long long lru_clock); long long lru_clock, int lru_multiplier);
#define LOOKUP_NONE 0 #define LOOKUP_NONE 0
#define LOOKUP_NOTOUCH (1<<0) #define LOOKUP_NOTOUCH (1<<0)
void dbAdd(redisDb *db, robj *key, robj *val); void dbAdd(redisDb *db, robj *key, robj *val);
......
...@@ -1220,6 +1220,14 @@ void xaddCommand(client *c) { ...@@ -1220,6 +1220,14 @@ void xaddCommand(client *c) {
return; return;
} }
/* Return ASAP if minimal ID (0-0) was given so we avoid possibly creating
* a new stream and have streamAppendItem fail, leaving an empty key in the
* database. */
if (id_given && id.ms == 0 && id.seq == 0) {
addReplyError(c,"The ID specified in XADD must be greater than 0-0");
return;
}
/* Lookup the stream at key. */ /* Lookup the stream at key. */
robj *o; robj *o;
stream *s; stream *s;
......
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