Unverified Commit 64c2508e authored by Salvatore Sanfilippo's avatar Salvatore Sanfilippo Committed by GitHub
Browse files

Merge branch 'unstable' into rm_get_server_info

parents 04233097 f1f259de
...@@ -533,6 +533,12 @@ int masterTryPartialResynchronization(client *c) { ...@@ -533,6 +533,12 @@ int masterTryPartialResynchronization(client *c) {
* has this state from the previous connection with the master. */ * has this state from the previous connection with the master. */
refreshGoodSlavesCount(); refreshGoodSlavesCount();
/* Fire the replica change modules event. */
moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE,
REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE,
NULL);
return C_OK; /* The caller can return, no full resync needed. */ return C_OK; /* The caller can return, no full resync needed. */
need_full_resync: need_full_resync:
...@@ -868,6 +874,10 @@ void putSlaveOnline(client *slave) { ...@@ -868,6 +874,10 @@ void putSlaveOnline(client *slave) {
return; return;
} }
refreshGoodSlavesCount(); refreshGoodSlavesCount();
/* Fire the replica change modules event. */
moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE,
REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE,
NULL);
serverLog(LL_NOTICE,"Synchronization with replica %s succeeded", serverLog(LL_NOTICE,"Synchronization with replica %s succeeded",
replicationGetSlaveName(slave)); replicationGetSlaveName(slave));
} }
...@@ -1542,11 +1552,11 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1542,11 +1552,11 @@ void readSyncBulkPayload(connection *conn) {
* We'll restore it when the RDB is received. */ * We'll restore it when the RDB is received. */
connBlock(conn); connBlock(conn);
connRecvTimeout(conn, server.repl_timeout*1000); connRecvTimeout(conn, server.repl_timeout*1000);
startLoading(server.repl_transfer_size); startLoading(server.repl_transfer_size, RDBFLAGS_REPLICATION);
if (rdbLoadRio(&rdb,&rsi,0) != C_OK) { if (rdbLoadRio(&rdb,RDBFLAGS_REPLICATION,&rsi) != C_OK) {
/* RDB loading failed. */ /* RDB loading failed. */
stopLoading(); stopLoading(0);
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Failed trying to load the MASTER synchronization DB " "Failed trying to load the MASTER synchronization DB "
"from socket"); "from socket");
...@@ -1567,7 +1577,7 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1567,7 +1577,7 @@ void readSyncBulkPayload(connection *conn) {
* gets promoted. */ * gets promoted. */
return; return;
} }
stopLoading(); stopLoading(1);
/* RDB loading succeeded if we reach this point. */ /* RDB loading succeeded if we reach this point. */
if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) { if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
...@@ -1614,7 +1624,7 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1614,7 +1624,7 @@ void readSyncBulkPayload(connection *conn) {
cancelReplicationHandshake(); cancelReplicationHandshake();
return; return;
} }
if (rdbLoad(server.rdb_filename,&rsi) != C_OK) { if (rdbLoad(server.rdb_filename,&rsi,RDBFLAGS_REPLICATION) != C_OK) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Failed trying to load the MASTER synchronization " "Failed trying to load the MASTER synchronization "
"DB from disk"); "DB from disk");
...@@ -1636,6 +1646,11 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1636,6 +1646,11 @@ void readSyncBulkPayload(connection *conn) {
server.repl_state = REPL_STATE_CONNECTED; server.repl_state = REPL_STATE_CONNECTED;
server.repl_down_since = 0; server.repl_down_since = 0;
/* Fire the master link modules event. */
moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
REDISMODULE_SUBEVENT_MASTER_LINK_UP,
NULL);
/* After a full resynchroniziation we use the replication ID and /* After a full resynchroniziation we use the replication ID and
* offset of the master. The secondary ID / offset are cleared since * offset of the master. The secondary ID / offset are cleared since
* we are starting a new history. */ * we are starting a new history. */
...@@ -2314,12 +2329,31 @@ void replicationSetMaster(char *ip, int port) { ...@@ -2314,12 +2329,31 @@ void replicationSetMaster(char *ip, int port) {
replicationDiscardCachedMaster(); replicationDiscardCachedMaster();
replicationCacheMasterUsingMyself(); replicationCacheMasterUsingMyself();
} }
/* Fire the role change modules event. */
moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED,
REDISMODULE_EVENT_REPLROLECHANGED_NOW_REPLICA,
NULL);
/* Fire the master link modules event. */
if (server.repl_state == REPL_STATE_CONNECTED)
moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
REDISMODULE_SUBEVENT_MASTER_LINK_DOWN,
NULL);
server.repl_state = REPL_STATE_CONNECT; server.repl_state = REPL_STATE_CONNECT;
} }
/* Cancel replication, setting the instance as a master itself. */ /* Cancel replication, setting the instance as a master itself. */
void replicationUnsetMaster(void) { void replicationUnsetMaster(void) {
if (server.masterhost == NULL) return; /* Nothing to do. */ if (server.masterhost == NULL) return; /* Nothing to do. */
/* Fire the master link modules event. */
if (server.repl_state == REPL_STATE_CONNECTED)
moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
REDISMODULE_SUBEVENT_MASTER_LINK_DOWN,
NULL);
sdsfree(server.masterhost); sdsfree(server.masterhost);
server.masterhost = NULL; server.masterhost = NULL;
/* When a slave is turned into a master, the current replication ID /* When a slave is turned into a master, the current replication ID
...@@ -2348,11 +2382,22 @@ void replicationUnsetMaster(void) { ...@@ -2348,11 +2382,22 @@ void replicationUnsetMaster(void) {
* starting from now. Otherwise the backlog will be freed after a * starting from now. Otherwise the backlog will be freed after a
* failover if slaves do not connect immediately. */ * failover if slaves do not connect immediately. */
server.repl_no_slaves_since = server.unixtime; server.repl_no_slaves_since = server.unixtime;
/* Fire the role change modules event. */
moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED,
REDISMODULE_EVENT_REPLROLECHANGED_NOW_MASTER,
NULL);
} }
/* This function is called when the slave lose the connection with the /* This function is called when the slave lose the connection with the
* master into an unexpected way. */ * master into an unexpected way. */
void replicationHandleMasterDisconnection(void) { void replicationHandleMasterDisconnection(void) {
/* Fire the master link modules event. */
if (server.repl_state == REPL_STATE_CONNECTED)
moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
REDISMODULE_SUBEVENT_MASTER_LINK_DOWN,
NULL);
server.master = NULL; server.master = NULL;
server.repl_state = REPL_STATE_CONNECT; server.repl_state = REPL_STATE_CONNECT;
server.repl_down_since = server.unixtime; server.repl_down_since = server.unixtime;
......
...@@ -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). */
for (int type = 0; type < 2; type++) {
retval = redisAsyncCommand(ri->link->cc, retval = redisAsyncCommand(ri->link->cc,
sentinelDiscardReplyCallback, ri, "%s KILL TYPE normal", sentinelDiscardReplyCallback, ri, "%s KILL TYPE %s",
sentinelInstanceMapCommand(ri,"CLIENT")); sentinelInstanceMapCommand(ri,"CLIENT"),
type == 0 ? "normal" : "pubsub");
if (retval == C_ERR) return retval; if (retval == C_ERR) return retval;
ri->link->pending_commands++; ri->link->pending_commands++;
}
retval = redisAsyncCommand(ri->link->cc, retval = redisAsyncCommand(ri->link->cc,
sentinelDiscardReplyCallback, ri, "%s", sentinelDiscardReplyCallback, ri, "%s",
......
...@@ -1691,7 +1691,6 @@ void databasesCron(void) { ...@@ -1691,7 +1691,6 @@ 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
...@@ -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. */
if (update_daylight_info) {
struct tm tm; struct tm tm;
time_t ut = server.unixtime; time_t ut = server.unixtime;
localtime_r(&ut,&tm); localtime_r(&ut,&tm);
server.daylight_active = tm.tm_isdst; 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
...@@ -2056,6 +2064,12 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { ...@@ -2056,6 +2064,12 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
server.rdb_bgsave_scheduled = 0; server.rdb_bgsave_scheduled = 0;
} }
/* Fire the cron loop modules event. */
RedisModuleCronLoopV1 ei = {REDISMODULE_CRON_LOOP_VERSION,server.hz};
moduleFireServerEvent(REDISMODULE_EVENT_CRON_LOOP,
0,
&ei);
server.cronloops++; server.cronloops++;
return 1000/server.hz; return 1000/server.hz;
} }
...@@ -2252,7 +2266,7 @@ void createSharedObjects(void) { ...@@ -2252,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();
...@@ -2279,6 +2293,7 @@ void initServerConfig(void) { ...@@ -2279,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;
...@@ -2730,6 +2745,7 @@ void resetServerStats(void) { ...@@ -2730,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;
...@@ -2771,6 +2787,7 @@ void initServer(void) { ...@@ -2771,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();
...@@ -2832,12 +2849,14 @@ void initServer(void) { ...@@ -2832,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);
...@@ -3238,10 +3257,13 @@ void preventCommandReplication(client *c) { ...@@ -3238,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) &&
...@@ -3259,7 +3281,8 @@ void call(client *c, int flags) { ...@@ -3259,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;
...@@ -3366,6 +3389,7 @@ void call(client *c, int flags) { ...@@ -3366,6 +3389,7 @@ void call(client *c, int flags) {
trackingRememberKeys(caller); trackingRememberKeys(caller);
} }
server.fixed_time_expire--;
server.stat_numcommands++; server.stat_numcommands++;
} }
...@@ -3682,6 +3706,9 @@ int prepareForShutdown(int flags) { ...@@ -3682,6 +3706,9 @@ int prepareForShutdown(int flags) {
} }
} }
/* Fire the shutdown modules event. */
moduleFireServerEvent(REDISMODULE_EVENT_SHUTDOWN,0,NULL);
/* Remove the pid file if possible and needed. */ /* Remove the pid file if possible and needed. */
if (server.daemonize || server.pidfile) { if (server.daemonize || server.pidfile) {
serverLog(LL_NOTICE,"Removing the pid file."); serverLog(LL_NOTICE,"Removing the pid file.");
...@@ -4244,6 +4271,7 @@ sds genRedisInfoString(const char *section) { ...@@ -4244,6 +4271,7 @@ sds genRedisInfoString(const 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"
...@@ -4271,6 +4299,7 @@ sds genRedisInfoString(const char *section) { ...@@ -4271,6 +4299,7 @@ sds genRedisInfoString(const 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,
...@@ -4767,7 +4796,7 @@ void loadDataFromDisk(void) { ...@@ -4767,7 +4796,7 @@ void loadDataFromDisk(void) {
serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000); serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000);
} else { } else {
rdbSaveInfo rsi = RDB_SAVE_INFO_INIT; rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
if (rdbLoad(server.rdb_filename,&rsi) == C_OK) { if (rdbLoad(server.rdb_filename,&rsi,RDBFLAGS_NONE) == C_OK) {
serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds", serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds",
(float)(ustime()-start)/1000000); (float)(ustime()-start)/1000000);
......
...@@ -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 */
...@@ -1602,6 +1606,7 @@ ssize_t rdbSaveModulesAux(rio *rdb, int when); ...@@ -1602,6 +1606,7 @@ ssize_t rdbSaveModulesAux(rio *rdb, int when);
int moduleAllDatatypesHandleErrors(); int moduleAllDatatypesHandleErrors();
sds modulesCollectInfo(sds info, const char *section, int for_crash_report, int sections); sds modulesCollectInfo(sds info, const char *section, int for_crash_report, int sections);
void moduleFireServerEvent(uint64_t eid, int subid, void *data); void moduleFireServerEvent(uint64_t eid, int subid, void *data);
void processModuleLoadingProgressEvent(int is_aof);
int moduleTryServeClientBlockedOnKey(client *c, robj *key); int moduleTryServeClientBlockedOnKey(client *c, robj *key);
void moduleUnblockClient(client *c); void moduleUnblockClient(client *c);
int moduleClientIsBlockedOnKeys(client *c); int moduleClientIsBlockedOnKeys(client *c);
...@@ -1834,10 +1839,12 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, ...@@ -1834,10 +1839,12 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData,
void rdbPipeWriteHandlerConnRemoved(struct connection *conn); void rdbPipeWriteHandlerConnRemoved(struct connection *conn);
/* Generic persistence functions */ /* Generic persistence functions */
void startLoadingFile(FILE* fp, char* filename); void startLoadingFile(FILE* fp, char* filename, int rdbflags);
void startLoading(size_t size); void startLoading(size_t size, int rdbflags);
void loadingProgress(off_t pos); void loadingProgress(off_t pos);
void stopLoading(void); void stopLoading(int success);
void startSaving(int rdbflags);
void stopSaving(int success);
#define DISK_ERROR_TYPE_AOF 1 /* Don't accept writes: AOF errors. */ #define DISK_ERROR_TYPE_AOF 1 /* Don't accept writes: AOF errors. */
#define DISK_ERROR_TYPE_RDB 2 /* Don't accept writes: RDB errors. */ #define DISK_ERROR_TYPE_RDB 2 /* Don't accept writes: RDB errors. */
...@@ -1846,7 +1853,6 @@ int writeCommandsDeniedByDiskError(void); ...@@ -1846,7 +1853,6 @@ int writeCommandsDeniedByDiskError(void);
/* RDB persistence */ /* RDB persistence */
#include "rdb.h" #include "rdb.h"
int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi);
void killRDBChild(void); void killRDBChild(void);
/* AOF persistence */ /* AOF persistence */
...@@ -1862,6 +1868,7 @@ void aofRewriteBufferReset(void); ...@@ -1862,6 +1868,7 @@ void aofRewriteBufferReset(void);
unsigned long aofRewriteBufferSize(void); unsigned long aofRewriteBufferSize(void);
ssize_t aofReadDiffFromParent(void); ssize_t aofReadDiffFromParent(void);
void killAppendOnlyChild(void); void killAppendOnlyChild(void);
void restartAOFAfterSYNC();
/* Child info */ /* Child info */
void openChildInfoPipe(void); void openChildInfoPipe(void);
...@@ -1996,7 +2003,7 @@ void populateCommandTable(void); ...@@ -1996,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);
...@@ -2082,10 +2089,11 @@ robj *lookupKeyWrite(redisDb *db, robj *key); ...@@ -2082,10 +2089,11 @@ robj *lookupKeyWrite(redisDb *db, robj *key);
robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply); robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply);
robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply); robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply);
robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags); robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags);
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);
void 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);
...@@ -2101,6 +2109,7 @@ robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o); ...@@ -2101,6 +2109,7 @@ robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o);
#define EMPTYDB_ASYNC (1<<0) /* Reclaim memory in another thread. */ #define EMPTYDB_ASYNC (1<<0) /* Reclaim memory in another thread. */
long long emptyDb(int dbnum, int flags, void(callback)(void*)); long long emptyDb(int dbnum, int flags, void(callback)(void*));
long long emptyDbGeneric(redisDb *dbarray, int dbnum, int flags, void(callback)(void*)); long long emptyDbGeneric(redisDb *dbarray, int dbnum, int flags, void(callback)(void*));
void flushAllDataAndResetRDB(int flags);
long long dbTotalServerKeyCount(); long long dbTotalServerKeyCount();
int selectDb(client *c, int id); int selectDb(client *c, int id);
......
...@@ -98,6 +98,7 @@ struct client; ...@@ -98,6 +98,7 @@ struct client;
stream *streamNew(void); stream *streamNew(void);
void freeStream(stream *s); void freeStream(stream *s);
unsigned long streamLength(const robj *subject);
size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end, size_t count, int rev, streamCG *group, streamConsumer *consumer, int flags, streamPropInfo *spi); size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end, size_t count, int rev, streamCG *group, streamConsumer *consumer, int flags, streamPropInfo *spi);
void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev); void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev);
int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields); int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields);
......
...@@ -621,7 +621,7 @@ void hincrbyfloatCommand(client *c) { ...@@ -621,7 +621,7 @@ void hincrbyfloatCommand(client *c) {
} }
char buf[MAX_LONG_DOUBLE_CHARS]; char buf[MAX_LONG_DOUBLE_CHARS];
int len = ld2string(buf,sizeof(buf),value,1); int len = ld2string(buf,sizeof(buf),value,LD_STR_HUMAN);
new = sdsnewlen(buf,len); new = sdsnewlen(buf,len);
hashTypeSet(o,c->argv[2]->ptr,new,HASH_SET_TAKE_VALUE); hashTypeSet(o,c->argv[2]->ptr,new,HASH_SET_TAKE_VALUE);
addReplyBulkCBuffer(c,buf,len); addReplyBulkCBuffer(c,buf,len);
......
...@@ -67,6 +67,12 @@ void freeStream(stream *s) { ...@@ -67,6 +67,12 @@ void freeStream(stream *s) {
zfree(s); zfree(s);
} }
/* Return the length of a stream. */
unsigned long streamLength(const robj *subject) {
stream *s = subject->ptr;
return s->length;
}
/* Generate the next stream item ID given the previous one. If the current /* Generate the next stream item ID given the previous one. If the current
* milliseconds Unix time is greater than the previous one, just use this * milliseconds Unix time is greater than the previous one, just use this
* as time part and start with sequence part of zero. Otherwise we use the * as time part and start with sequence part of zero. Otherwise we use the
...@@ -173,9 +179,19 @@ int streamCompareID(streamID *a, streamID *b) { ...@@ -173,9 +179,19 @@ int streamCompareID(streamID *a, streamID *b) {
* C_ERR if an ID was given via 'use_id', but adding it failed since the * C_ERR if an ID was given via 'use_id', but adding it failed since the
* current top ID is greater or equal. */ * current top ID is greater or equal. */
int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id) { int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id) {
/* If an ID was given, check that it's greater than the last entry ID
* or return an error. */ /* Generate the new entry ID. */
if (use_id && streamCompareID(use_id,&s->last_id) <= 0) return C_ERR; streamID id;
if (use_id)
id = *use_id;
else
streamNextID(&s->last_id,&id);
/* Check that the new ID is greater than the last entry ID
* or return an error. Automatically generated IDs might
* overflow (and wrap-around) when incrementing the sequence
part. */
if (streamCompareID(&id,&s->last_id) <= 0) return C_ERR;
/* Add the new entry. */ /* Add the new entry. */
raxIterator ri; raxIterator ri;
...@@ -192,13 +208,6 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_ ...@@ -192,13 +208,6 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
} }
raxStop(&ri); raxStop(&ri);
/* Generate the new entry ID. */
streamID id;
if (use_id)
id = *use_id;
else
streamNextID(&s->last_id,&id);
/* We have to add the key into the radix tree in lexicographic order, /* We have to add the key into the radix tree in lexicographic order,
* to do so we consider the ID as a single 128 bit number written in * to do so we consider the ID as a single 128 bit number written in
* big endian, so that the most significant bytes are the first ones. */ * big endian, so that the most significant bytes are the first ones. */
...@@ -1197,6 +1206,14 @@ void xaddCommand(client *c) { ...@@ -1197,6 +1206,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;
......
...@@ -551,15 +551,17 @@ int d2string(char *buf, size_t len, double value) { ...@@ -551,15 +551,17 @@ int d2string(char *buf, size_t len, double value) {
return len; return len;
} }
/* Convert a long double into a string. If humanfriendly is non-zero /* Create a string object from a long double.
* it does not use exponential format and trims trailing zeroes at the end, * If mode is humanfriendly it does not use exponential format and trims trailing
* however this results in loss of precision. Otherwise exp format is used * zeroes at the end (may result in loss of precision).
* and the output of snprintf() is not modified. * If mode is default exp format is used and the output of snprintf()
* is not modified (may result in loss of precision).
* If mode is hex hexadecimal format is used (no loss of precision)
* *
* The function returns the length of the string or zero if there was not * The function returns the length of the string or zero if there was not
* enough buffer room to store it. */ * enough buffer room to store it. */
int ld2string(char *buf, size_t len, long double value, int humanfriendly) { int ld2string(char *buf, size_t len, long double value, ld2string_mode mode) {
size_t l; size_t l = 0;
if (isinf(value)) { if (isinf(value)) {
/* Libc in odd systems (Hi Solaris!) will format infinite in a /* Libc in odd systems (Hi Solaris!) will format infinite in a
...@@ -572,13 +574,23 @@ int ld2string(char *buf, size_t len, long double value, int humanfriendly) { ...@@ -572,13 +574,23 @@ int ld2string(char *buf, size_t len, long double value, int humanfriendly) {
memcpy(buf,"-inf",4); memcpy(buf,"-inf",4);
l = 4; l = 4;
} }
} else if (humanfriendly) { } else {
switch (mode) {
case LD_STR_AUTO:
l = snprintf(buf,len,"%.17Lg",value);
if (l+1 > len) return 0; /* No room. */
break;
case LD_STR_HEX:
l = snprintf(buf,len,"%La",value);
if (l+1 > len) return 0; /* No room. */
break;
case LD_STR_HUMAN:
/* We use 17 digits precision since with 128 bit floats that precision /* We use 17 digits precision since with 128 bit floats that precision
* after rounding is able to represent most small decimal numbers in a * after rounding is able to represent most small decimal numbers in a
* way that is "non surprising" for the user (that is, most small * way that is "non surprising" for the user (that is, most small
* decimal numbers will be represented in a way that when converted * decimal numbers will be represented in a way that when converted
* back into a string are exactly the same as what the user typed.) */ * back into a string are exactly the same as what the user typed.) */
l = snprintf(buf,len,"%.17Lf", value); l = snprintf(buf,len,"%.17Lf",value);
if (l+1 > len) return 0; /* No room. */ if (l+1 > len) return 0; /* No room. */
/* Now remove trailing zeroes after the '.' */ /* Now remove trailing zeroes after the '.' */
if (strchr(buf,'.') != NULL) { if (strchr(buf,'.') != NULL) {
...@@ -589,9 +601,9 @@ int ld2string(char *buf, size_t len, long double value, int humanfriendly) { ...@@ -589,9 +601,9 @@ int ld2string(char *buf, size_t len, long double value, int humanfriendly) {
} }
if (*p == '.') l--; if (*p == '.') l--;
} }
} else { break;
l = snprintf(buf,len,"%.17Lg", value); default: return 0; /* Invalid mode. */
if (l+1 > len) return 0; /* No room. */ }
} }
buf[l] = '\0'; buf[l] = '\0';
return l; return l;
......
...@@ -38,6 +38,13 @@ ...@@ -38,6 +38,13 @@
* This should be the size of the buffer given to ld2string */ * This should be the size of the buffer given to ld2string */
#define MAX_LONG_DOUBLE_CHARS 5*1024 #define MAX_LONG_DOUBLE_CHARS 5*1024
/* long double to string convertion options */
typedef enum {
LD_STR_AUTO, /* %.17Lg */
LD_STR_HUMAN, /* %.17Lf + Trimming of trailing zeros */
LD_STR_HEX /* %La */
} ld2string_mode;
int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase); int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase);
int stringmatch(const char *p, const char *s, int nocase); int stringmatch(const char *p, const char *s, int nocase);
int stringmatchlen_fuzz_test(void); int stringmatchlen_fuzz_test(void);
...@@ -51,7 +58,7 @@ int string2l(const char *s, size_t slen, long *value); ...@@ -51,7 +58,7 @@ int string2l(const char *s, size_t slen, long *value);
int string2ld(const char *s, size_t slen, long double *dp); int string2ld(const char *s, size_t slen, long double *dp);
int string2d(const char *s, size_t slen, double *dp); int string2d(const char *s, size_t slen, double *dp);
int d2string(char *buf, size_t len, double value); int d2string(char *buf, size_t len, double value);
int ld2string(char *buf, size_t len, long double value, int humanfriendly); int ld2string(char *buf, size_t len, long double value, ld2string_mode mode);
sds getAbsolutePath(char *filename); sds getAbsolutePath(char *filename);
unsigned long getTimeZone(void); unsigned long getTimeZone(void);
int pathIsBaseName(char *path); int pathIsBaseName(char *path);
......
...@@ -18,7 +18,10 @@ TEST_MODULES = \ ...@@ -18,7 +18,10 @@ TEST_MODULES = \
infotest.so \ infotest.so \
propagate.so \ propagate.so \
misc.so \ misc.so \
hooks.so hooks.so \
blockonkeys.so \
scan.so \
datatype.so
.PHONY: all .PHONY: all
......
#define REDISMODULE_EXPERIMENTAL_API
#include "redismodule.h"
#include <string.h>
#include <assert.h>
#include <unistd.h>
#define LIST_SIZE 1024
typedef struct {
long long list[LIST_SIZE];
long long length;
} fsl_t; /* Fixed-size list */
static RedisModuleType *fsltype = NULL;
fsl_t *fsl_type_create() {
fsl_t *o;
o = RedisModule_Alloc(sizeof(*o));
o->length = 0;
return o;
}
void fsl_type_free(fsl_t *o) {
RedisModule_Free(o);
}
/* ========================== "fsltype" type methods ======================= */
void *fsl_rdb_load(RedisModuleIO *rdb, int encver) {
if (encver != 0) {
return NULL;
}
fsl_t *fsl = fsl_type_create();
fsl->length = RedisModule_LoadUnsigned(rdb);
for (long long i = 0; i < fsl->length; i++)
fsl->list[i] = RedisModule_LoadSigned(rdb);
return fsl;
}
void fsl_rdb_save(RedisModuleIO *rdb, void *value) {
fsl_t *fsl = value;
RedisModule_SaveUnsigned(rdb,fsl->length);
for (long long i = 0; i < fsl->length; i++)
RedisModule_SaveSigned(rdb, fsl->list[i]);
}
void fsl_aofrw(RedisModuleIO *aof, RedisModuleString *key, void *value) {
fsl_t *fsl = value;
for (long long i = 0; i < fsl->length; i++)
RedisModule_EmitAOF(aof, "FSL.PUSH","sl", key, fsl->list[i]);
}
void fsl_free(void *value) {
fsl_type_free(value);
}
/* ========================== helper methods ======================= */
int get_fsl(RedisModuleCtx *ctx, RedisModuleString *keyname, int mode, int create, fsl_t **fsl, int reply_on_failure) {
RedisModuleKey *key = RedisModule_OpenKey(ctx, keyname, mode);
int type = RedisModule_KeyType(key);
if (type != REDISMODULE_KEYTYPE_EMPTY && RedisModule_ModuleTypeGetType(key) != fsltype) {
RedisModule_CloseKey(key);
if (reply_on_failure)
RedisModule_ReplyWithError(ctx, REDISMODULE_ERRORMSG_WRONGTYPE);
return 0;
}
/* Create an empty value object if the key is currently empty. */
if (type == REDISMODULE_KEYTYPE_EMPTY) {
if (!create) {
/* Key is empty but we cannot create */
RedisModule_CloseKey(key);
*fsl = NULL;
return 1;
}
*fsl = fsl_type_create();
RedisModule_ModuleTypeSetValue(key, fsltype, *fsl);
} else {
*fsl = RedisModule_ModuleTypeGetValue(key);
}
RedisModule_CloseKey(key);
return 1;
}
/* ========================== commands ======================= */
/* FSL.PUSH <key> <int> - Push an integer to the fixed-size list (to the right).
* It must be greater than the element in the head of the list. */
int fsl_push(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 3)
return RedisModule_WrongArity(ctx);
long long ele;
if (RedisModule_StringToLongLong(argv[2],&ele) != REDISMODULE_OK)
return RedisModule_ReplyWithError(ctx,"ERR invalid integer");
fsl_t *fsl;
if (!get_fsl(ctx, argv[1], REDISMODULE_WRITE, 1, &fsl, 1))
return REDISMODULE_OK;
if (fsl->length == LIST_SIZE)
return RedisModule_ReplyWithError(ctx,"ERR list is full");
if (fsl->length != 0 && fsl->list[fsl->length-1] >= ele)
return RedisModule_ReplyWithError(ctx,"ERR new element has to be greater than the head element");
fsl->list[fsl->length++] = ele;
if (fsl->length >= 2)
RedisModule_SignalKeyAsReady(ctx, argv[1]);
return RedisModule_ReplyWithSimpleString(ctx, "OK");
}
int bpop2_reply_callback(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
RedisModuleString *keyname = RedisModule_GetBlockedClientReadyKey(ctx);
fsl_t *fsl;
if (!get_fsl(ctx, keyname, REDISMODULE_READ, 0, &fsl, 0))
return REDISMODULE_ERR;
if (!fsl || fsl->length < 2)
return REDISMODULE_ERR;
RedisModule_ReplyWithArray(ctx, 2);
RedisModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);
RedisModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);
return REDISMODULE_OK;
}
int bpop2_timeout_callback(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
return RedisModule_ReplyWithSimpleString(ctx, "Request timedout");
}
/* FSL.BPOP2 <key> <timeout> - Block clients until list has two or more elements.
* When that happens, unblock client and pop the last two elements (from the right). */
int fsl_bpop2(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 3)
return RedisModule_WrongArity(ctx);
long long timeout;
if (RedisModule_StringToLongLong(argv[2],&timeout) != REDISMODULE_OK || timeout < 0)
return RedisModule_ReplyWithError(ctx,"ERR invalid timeout");
fsl_t *fsl;
if (!get_fsl(ctx, argv[1], REDISMODULE_READ, 0, &fsl, 1))
return REDISMODULE_OK;
if (!fsl || fsl->length < 2) {
/* Key is empty or has <2 elements, we must block */
RedisModule_BlockClientOnKeys(ctx, bpop2_reply_callback, bpop2_timeout_callback,
NULL, timeout, &argv[1], 1, NULL);
} else {
RedisModule_ReplyWithArray(ctx, 2);
RedisModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);
RedisModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);
}
return REDISMODULE_OK;
}
int bpopgt_reply_callback(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
RedisModuleString *keyname = RedisModule_GetBlockedClientReadyKey(ctx);
long long gt = (long long)RedisModule_GetBlockedClientPrivateData(ctx);
fsl_t *fsl;
if (!get_fsl(ctx, keyname, REDISMODULE_READ, 0, &fsl, 0))
return REDISMODULE_ERR;
if (!fsl || fsl->list[fsl->length-1] <= gt)
return REDISMODULE_ERR;
RedisModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);
return REDISMODULE_OK;
}
int bpopgt_timeout_callback(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
return RedisModule_ReplyWithSimpleString(ctx, "Request timedout");
}
void bpopgt_free_privdata(RedisModuleCtx *ctx, void *privdata) {
/* Nothing to do because privdata is actually a 'long long',
* not a pointer to the heap */
REDISMODULE_NOT_USED(ctx);
REDISMODULE_NOT_USED(privdata);
}
/* FSL.BPOPGT <key> <gt> <timeout> - Block clients until list has an element greater than <gt>.
* When that happens, unblock client and pop the last element (from the right). */
int fsl_bpopgt(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 4)
return RedisModule_WrongArity(ctx);
long long gt;
if (RedisModule_StringToLongLong(argv[2],&gt) != REDISMODULE_OK)
return RedisModule_ReplyWithError(ctx,"ERR invalid integer");
long long timeout;
if (RedisModule_StringToLongLong(argv[3],&timeout) != REDISMODULE_OK || timeout < 0)
return RedisModule_ReplyWithError(ctx,"ERR invalid timeout");
fsl_t *fsl;
if (!get_fsl(ctx, argv[1], REDISMODULE_READ, 0, &fsl, 1))
return REDISMODULE_OK;
if (!fsl || fsl->list[fsl->length-1] <= gt) {
/* Key is empty or has <2 elements, we must block */
RedisModule_BlockClientOnKeys(ctx, bpopgt_reply_callback, bpopgt_timeout_callback,
bpopgt_free_privdata, timeout, &argv[1], 1, (void*)gt);
} else {
RedisModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);
}
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
if (RedisModule_Init(ctx, "blockonkeys", 1, REDISMODULE_APIVER_1)== REDISMODULE_ERR)
return REDISMODULE_ERR;
RedisModuleTypeMethods tm = {
.version = REDISMODULE_TYPE_METHOD_VERSION,
.rdb_load = fsl_rdb_load,
.rdb_save = fsl_rdb_save,
.aof_rewrite = fsl_aofrw,
.mem_usage = NULL,
.free = fsl_free,
.digest = NULL
};
fsltype = RedisModule_CreateDataType(ctx, "fsltype_t", 0, &tm);
if (fsltype == NULL)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"fsl.push",fsl_push,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"fsl.bpop2",fsl_bpop2,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"fsl.bpopgt",fsl_bpopgt,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}
/* This module current tests a small subset but should be extended in the future
* for general ModuleDataType coverage.
*/
#include "redismodule.h"
static RedisModuleType *datatype = NULL;
typedef struct {
long long intval;
RedisModuleString *strval;
} DataType;
static void *datatype_load(RedisModuleIO *io, int encver) {
(void) encver;
int intval = RedisModule_LoadSigned(io);
if (RedisModule_IsIOError(io)) return NULL;
RedisModuleString *strval = RedisModule_LoadString(io);
if (RedisModule_IsIOError(io)) return NULL;
DataType *dt = (DataType *) RedisModule_Alloc(sizeof(DataType));
dt->intval = intval;
dt->strval = strval;
return dt;
}
static void datatype_save(RedisModuleIO *io, void *value) {
DataType *dt = (DataType *) value;
RedisModule_SaveSigned(io, dt->intval);
RedisModule_SaveString(io, dt->strval);
}
static void datatype_free(void *value) {
if (value) {
DataType *dt = (DataType *) value;
if (dt->strval) RedisModule_FreeString(NULL, dt->strval);
RedisModule_Free(dt);
}
}
static int datatype_set(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 4) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
long long intval;
if (RedisModule_StringToLongLong(argv[2], &intval) != REDISMODULE_OK) {
RedisModule_ReplyWithError(ctx, "Invalid integr value");
return REDISMODULE_OK;
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
DataType *dt = RedisModule_Calloc(sizeof(DataType), 1);
dt->intval = intval;
dt->strval = argv[3];
RedisModule_RetainString(ctx, dt->strval);
RedisModule_ModuleTypeSetValue(key, datatype, dt);
RedisModule_CloseKey(key);
RedisModule_ReplyWithSimpleString(ctx, "OK");
return REDISMODULE_OK;
}
static int datatype_restore(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 3) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
DataType *dt = RedisModule_LoadDataTypeFromString(argv[2], datatype);
if (!dt) {
RedisModule_ReplyWithError(ctx, "Invalid data");
return REDISMODULE_OK;
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
RedisModule_ModuleTypeSetValue(key, datatype, dt);
RedisModule_CloseKey(key);
RedisModule_ReplyWithSimpleString(ctx, "OK");
return REDISMODULE_OK;
}
static int datatype_get(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 2) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_READ);
DataType *dt = RedisModule_ModuleTypeGetValue(key);
RedisModule_CloseKey(key);
RedisModule_ReplyWithArray(ctx, 2);
RedisModule_ReplyWithLongLong(ctx, dt->intval);
RedisModule_ReplyWithString(ctx, dt->strval);
return REDISMODULE_OK;
}
static int datatype_dump(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 2) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_READ);
DataType *dt = RedisModule_ModuleTypeGetValue(key);
RedisModule_CloseKey(key);
RedisModuleString *reply = RedisModule_SaveDataTypeToString(ctx, dt, datatype);
if (!reply) {
RedisModule_ReplyWithError(ctx, "Failed to save");
return REDISMODULE_OK;
}
RedisModule_ReplyWithString(ctx, reply);
RedisModule_FreeString(ctx, reply);
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
if (RedisModule_Init(ctx,"datatype",1,REDISMODULE_APIVER_1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
RedisModule_SetModuleOptions(ctx, REDISMODULE_OPTIONS_HANDLE_IO_ERRORS);
RedisModuleTypeMethods datatype_methods = {
.version = REDISMODULE_TYPE_METHOD_VERSION,
.rdb_load = datatype_load,
.rdb_save = datatype_save,
.free = datatype_free,
};
datatype = RedisModule_CreateDataType(ctx, "test___dt", 1, &datatype_methods);
if (datatype == NULL)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"datatype.set", datatype_set,"deny-oom",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"datatype.get", datatype_get,"",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"datatype.restore", datatype_restore,"deny-oom",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"datatype.dump", datatype_dump,"",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}
...@@ -30,36 +30,227 @@ ...@@ -30,36 +30,227 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#define REDISMODULE_EXPERIMENTAL_API
#include "redismodule.h" #include "redismodule.h"
#include <stdio.h>
#include <string.h>
/* We need to store events to be able to test and see what we got, and we can't
* store them in the key-space since that would mess up rdb loading (duplicates)
* and be lost of flushdb. */
RedisModuleDict *event_log = NULL;
typedef struct EventElement {
long count;
RedisModuleString *last_val_string;
long last_val_int;
} EventElement;
void LogStringEvent(RedisModuleCtx *ctx, const char* keyname, const char* data) {
EventElement *event = RedisModule_DictGetC(event_log, (void*)keyname, strlen(keyname), NULL);
if (!event) {
event = RedisModule_Alloc(sizeof(EventElement));
memset(event, 0, sizeof(EventElement));
RedisModule_DictSetC(event_log, (void*)keyname, strlen(keyname), event);
}
if (event->last_val_string) RedisModule_FreeString(ctx, event->last_val_string);
event->last_val_string = RedisModule_CreateString(ctx, data, strlen(data));
event->count++;
}
void LogNumericEvent(RedisModuleCtx *ctx, const char* keyname, long data) {
REDISMODULE_NOT_USED(ctx);
EventElement *event = RedisModule_DictGetC(event_log, (void*)keyname, strlen(keyname), NULL);
if (!event) {
event = RedisModule_Alloc(sizeof(EventElement));
memset(event, 0, sizeof(EventElement));
RedisModule_DictSetC(event_log, (void*)keyname, strlen(keyname), event);
}
event->last_val_int = data;
event->count++;
}
void FreeEvent(RedisModuleCtx *ctx, EventElement *event) {
if (event->last_val_string)
RedisModule_FreeString(ctx, event->last_val_string);
RedisModule_Free(event);
}
int cmdEventCount(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
if (argc != 2){
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
EventElement *event = RedisModule_DictGet(event_log, argv[1], NULL);
RedisModule_ReplyWithLongLong(ctx, event? event->count: 0);
return REDISMODULE_OK;
}
int cmdEventLast(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
if (argc != 2){
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
EventElement *event = RedisModule_DictGet(event_log, argv[1], NULL);
if (event && event->last_val_string)
RedisModule_ReplyWithString(ctx, event->last_val_string);
else if (event)
RedisModule_ReplyWithLongLong(ctx, event->last_val_int);
else
RedisModule_ReplyWithNull(ctx);
return REDISMODULE_OK;
}
void clearEvents(RedisModuleCtx *ctx)
{
RedisModuleString *key;
EventElement *event;
RedisModuleDictIter *iter = RedisModule_DictIteratorStart(event_log, "^", NULL);
while((key = RedisModule_DictNext(ctx, iter, (void**)&event)) != NULL) {
event->count = 0;
event->last_val_int = 0;
if (event->last_val_string) RedisModule_FreeString(ctx, event->last_val_string);
event->last_val_string = NULL;
RedisModule_DictDel(event_log, key, NULL);
RedisModule_Free(event);
}
RedisModule_DictIteratorStop(iter);
}
int cmdEventsClear(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
REDISMODULE_NOT_USED(argc);
REDISMODULE_NOT_USED(argv);
clearEvents(ctx);
return REDISMODULE_OK;
}
/* Client state change callback. */ /* Client state change callback. */
void clientChangeCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data) void clientChangeCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{ {
REDISMODULE_NOT_USED(ctx);
REDISMODULE_NOT_USED(e); REDISMODULE_NOT_USED(e);
RedisModuleClientInfo *ci = data; RedisModuleClientInfo *ci = data;
char *keyname = (sub == REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED) ? char *keyname = (sub == REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED) ?
"connected" : "disconnected"; "client-connected" : "client-disconnected";
RedisModuleCallReply *reply; LogNumericEvent(ctx, keyname, ci->id);
RedisModule_SelectDb(ctx,9);
reply = RedisModule_Call(ctx,"RPUSH","cl",keyname,(long)ci->id);
RedisModule_FreeCallReply(reply);
} }
void flushdbCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data) void flushdbCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{ {
REDISMODULE_NOT_USED(ctx);
REDISMODULE_NOT_USED(e); REDISMODULE_NOT_USED(e);
RedisModuleFlushInfo *fi = data; RedisModuleFlushInfo *fi = data;
char *keyname = (sub == REDISMODULE_SUBEVENT_FLUSHDB_START) ? char *keyname = (sub == REDISMODULE_SUBEVENT_FLUSHDB_START) ?
"flush-start" : "flush-end"; "flush-start" : "flush-end";
RedisModuleCallReply *reply; LogNumericEvent(ctx, keyname, fi->dbnum);
RedisModule_SelectDb(ctx,9); }
reply = RedisModule_Call(ctx,"RPUSH","cl",keyname,(long)fi->dbnum);
RedisModule_FreeCallReply(reply); void roleChangeCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{
REDISMODULE_NOT_USED(e);
REDISMODULE_NOT_USED(data);
RedisModuleReplicationInfo *ri = data;
char *keyname = (sub == REDISMODULE_EVENT_REPLROLECHANGED_NOW_MASTER) ?
"role-master" : "role-replica";
LogStringEvent(ctx, keyname, ri->masterhost);
}
void replicationChangeCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{
REDISMODULE_NOT_USED(e);
REDISMODULE_NOT_USED(data);
char *keyname = (sub == REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE) ?
"replica-online" : "replica-offline";
LogNumericEvent(ctx, keyname, 0);
}
void rasterLinkChangeCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{
REDISMODULE_NOT_USED(e);
REDISMODULE_NOT_USED(data);
char *keyname = (sub == REDISMODULE_SUBEVENT_MASTER_LINK_UP) ?
"masterlink-up" : "masterlink-down";
LogNumericEvent(ctx, keyname, 0);
}
void persistenceCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{
REDISMODULE_NOT_USED(e);
REDISMODULE_NOT_USED(data);
char *keyname = NULL;
switch (sub) {
case REDISMODULE_SUBEVENT_PERSISTENCE_RDB_START: keyname = "persistence-rdb-start"; break;
case REDISMODULE_SUBEVENT_PERSISTENCE_AOF_START: keyname = "persistence-aof-start"; break;
case REDISMODULE_SUBEVENT_PERSISTENCE_SYNC_RDB_START: keyname = "persistence-syncrdb-start"; break;
case REDISMODULE_SUBEVENT_PERSISTENCE_ENDED: keyname = "persistence-end"; break;
case REDISMODULE_SUBEVENT_PERSISTENCE_FAILED: keyname = "persistence-failed"; break;
}
/* modifying the keyspace from the fork child is not an option, using log instead */
RedisModule_Log(ctx, "warning", "module-event-%s", keyname);
if (sub == REDISMODULE_SUBEVENT_PERSISTENCE_SYNC_RDB_START)
LogNumericEvent(ctx, keyname, 0);
}
void loadingCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{
REDISMODULE_NOT_USED(e);
REDISMODULE_NOT_USED(data);
char *keyname = NULL;
switch (sub) {
case REDISMODULE_SUBEVENT_LOADING_RDB_START: keyname = "loading-rdb-start"; break;
case REDISMODULE_SUBEVENT_LOADING_AOF_START: keyname = "loading-aof-start"; break;
case REDISMODULE_SUBEVENT_LOADING_REPL_START: keyname = "loading-repl-start"; break;
case REDISMODULE_SUBEVENT_LOADING_ENDED: keyname = "loading-end"; break;
case REDISMODULE_SUBEVENT_LOADING_FAILED: keyname = "loading-failed"; break;
}
LogNumericEvent(ctx, keyname, 0);
}
void loadingProgressCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{
REDISMODULE_NOT_USED(e);
RedisModuleLoadingProgress *ei = data;
char *keyname = (sub == REDISMODULE_SUBEVENT_LOADING_PROGRESS_RDB) ?
"loading-progress-rdb" : "loading-progress-aof";
LogNumericEvent(ctx, keyname, ei->progress);
}
void shutdownCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{
REDISMODULE_NOT_USED(e);
REDISMODULE_NOT_USED(data);
REDISMODULE_NOT_USED(sub);
RedisModule_Log(ctx, "warning", "module-event-%s", "shutdown");
}
void cronLoopCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{
REDISMODULE_NOT_USED(e);
REDISMODULE_NOT_USED(sub);
RedisModuleCronLoop *ei = data;
LogNumericEvent(ctx, "cron-loop", ei->hz);
}
void moduleChangeCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)
{
REDISMODULE_NOT_USED(e);
RedisModuleModuleChange *ei = data;
char *keyname = (sub == REDISMODULE_SUBEVENT_MODULE_LOADED) ?
"module-loaded" : "module-unloaded";
LogStringEvent(ctx, keyname, ei->module_name);
} }
/* This function must be present on each Redis module. It is used in order to /* This function must be present on each Redis module. It is used in order to
...@@ -71,9 +262,50 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -71,9 +262,50 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
if (RedisModule_Init(ctx,"testhook",1,REDISMODULE_APIVER_1) if (RedisModule_Init(ctx,"testhook",1,REDISMODULE_APIVER_1)
== REDISMODULE_ERR) return REDISMODULE_ERR; == REDISMODULE_ERR) return REDISMODULE_ERR;
/* replication related hooks */
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_ReplicationRoleChanged, roleChangeCallback);
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_ReplicaChange, replicationChangeCallback);
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_MasterLinkChange, rasterLinkChangeCallback);
/* persistence related hooks */
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_Persistence, persistenceCallback);
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_Loading, loadingCallback);
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_LoadingProgress, loadingProgressCallback);
/* other hooks */
RedisModule_SubscribeToServerEvent(ctx, RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_ClientChange, clientChangeCallback); RedisModuleEvent_ClientChange, clientChangeCallback);
RedisModule_SubscribeToServerEvent(ctx, RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_FlushDB, flushdbCallback); RedisModuleEvent_FlushDB, flushdbCallback);
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_Shutdown, shutdownCallback);
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_CronLoop, cronLoopCallback);
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_ModuleChange, moduleChangeCallback);
event_log = RedisModule_CreateDict(ctx);
if (RedisModule_CreateCommand(ctx,"hooks.event_count", cmdEventCount,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"hooks.event_last", cmdEventLast,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"hooks.clear", cmdEventsClear,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK; return REDISMODULE_OK;
} }
int RedisModule_OnUnload(RedisModuleCtx *ctx) {
clearEvents(ctx);
RedisModule_FreeDict(ctx, event_log);
event_log = NULL;
return REDISMODULE_OK;
}
...@@ -6,6 +6,8 @@ ...@@ -6,6 +6,8 @@
#include <unistd.h> #include <unistd.h>
#include <errno.h> #include <errno.h>
#define UNUSED(x) (void)(x)
int test_call_generic(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) int test_call_generic(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{ {
if (argc<2) { if (argc<2) {
...@@ -40,6 +42,146 @@ int test_call_info(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -40,6 +42,146 @@ int test_call_info(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
return REDISMODULE_OK; return REDISMODULE_OK;
} }
int test_ld_conv(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
UNUSED(argv);
UNUSED(argc);
long double ld = 0.00000000000000001L;
const char *ldstr = "0.00000000000000001";
RedisModuleString *s1 = RedisModule_CreateStringFromLongDouble(ctx, ld, 1);
RedisModuleString *s2 =
RedisModule_CreateString(ctx, ldstr, strlen(ldstr));
if (RedisModule_StringCompare(s1, s2) != 0) {
char err[4096];
snprintf(err, 4096,
"Failed to convert long double to string ('%s' != '%s')",
RedisModule_StringPtrLen(s1, NULL),
RedisModule_StringPtrLen(s2, NULL));
RedisModule_ReplyWithError(ctx, err);
goto final;
}
long double ld2 = 0;
if (RedisModule_StringToLongDouble(s2, &ld2) == REDISMODULE_ERR) {
RedisModule_ReplyWithError(ctx,
"Failed to convert string to long double");
goto final;
}
if (ld2 != ld) {
char err[4096];
snprintf(err, 4096,
"Failed to convert string to long double (%.40Lf != %.40Lf)",
ld2,
ld);
RedisModule_ReplyWithError(ctx, err);
goto final;
}
RedisModule_ReplyWithLongDouble(ctx, ld2);
final:
RedisModule_FreeString(ctx, s1);
RedisModule_FreeString(ctx, s2);
return REDISMODULE_OK;
}
int test_flushall(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
RedisModule_ResetDataset(1, 0);
RedisModule_ReplyWithCString(ctx, "Ok");
return REDISMODULE_OK;
}
int test_dbsize(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
long long ll = RedisModule_DbSize(ctx);
RedisModule_ReplyWithLongLong(ctx, ll);
return REDISMODULE_OK;
}
int test_randomkey(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
RedisModuleString *str = RedisModule_RandomKey(ctx);
RedisModule_ReplyWithString(ctx, str);
RedisModule_FreeString(ctx, str);
return REDISMODULE_OK;
}
RedisModuleKey *open_key_or_reply(RedisModuleCtx *ctx, RedisModuleString *keyname, int mode) {
RedisModuleKey *key = RedisModule_OpenKey(ctx, keyname, mode);
if (!key) {
RedisModule_ReplyWithError(ctx, "key not found");
return NULL;
}
return key;
}
int test_getlru(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
if (argc<2) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleKey *key = open_key_or_reply(ctx, argv[1], REDISMODULE_READ|REDISMODULE_OPEN_KEY_NOTOUCH);
mstime_t lru;
RedisModule_GetLRU(key, &lru);
RedisModule_ReplyWithLongLong(ctx, lru);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int test_setlru(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
if (argc<3) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleKey *key = open_key_or_reply(ctx, argv[1], REDISMODULE_READ|REDISMODULE_OPEN_KEY_NOTOUCH);
mstime_t lru;
if (RedisModule_StringToLongLong(argv[2], &lru) != REDISMODULE_OK) {
RedisModule_ReplyWithError(ctx, "invalid idle time");
return REDISMODULE_OK;
}
int was_set = RedisModule_SetLRU(key, lru)==REDISMODULE_OK;
RedisModule_ReplyWithLongLong(ctx, was_set);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int test_getlfu(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
if (argc<2) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleKey *key = open_key_or_reply(ctx, argv[1], REDISMODULE_READ|REDISMODULE_OPEN_KEY_NOTOUCH);
mstime_t lfu;
RedisModule_GetLFU(key, &lfu);
RedisModule_ReplyWithLongLong(ctx, lfu);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int test_setlfu(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
if (argc<3) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleKey *key = open_key_or_reply(ctx, argv[1], REDISMODULE_READ|REDISMODULE_OPEN_KEY_NOTOUCH);
mstime_t lfu;
if (RedisModule_StringToLongLong(argv[2], &lfu) != REDISMODULE_OK) {
RedisModule_ReplyWithError(ctx, "invalid freq");
return REDISMODULE_OK;
}
int was_set = RedisModule_SetLFU(key, lfu)==REDISMODULE_OK;
RedisModule_ReplyWithLongLong(ctx, was_set);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) { int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv); REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc); REDISMODULE_NOT_USED(argc);
...@@ -50,6 +192,22 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -50,6 +192,22 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.call_info", test_call_info,"",0,0,0) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"test.call_info", test_call_info,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.ld_conversion", test_ld_conv, "",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.flushall", test_flushall,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.dbsize", test_dbsize,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.randomkey", test_randomkey,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.setlru", test_setlru,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.getlru", test_getlru,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.setlfu", test_setlfu,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.getlfu", test_getlfu,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK; return REDISMODULE_OK;
} }
#include "redismodule.h"
#include <string.h>
#include <assert.h>
#include <unistd.h>
typedef struct {
size_t nkeys;
} scan_strings_pd;
void scan_strings_callback(RedisModuleCtx *ctx, RedisModuleString* keyname, RedisModuleKey* key, void *privdata) {
scan_strings_pd* pd = privdata;
int was_opened = 0;
if (!key) {
key = RedisModule_OpenKey(ctx, keyname, REDISMODULE_READ);
was_opened = 1;
}
if (RedisModule_KeyType(key) == REDISMODULE_KEYTYPE_STRING) {
size_t len;
char * data = RedisModule_StringDMA(key, &len, REDISMODULE_READ);
RedisModule_ReplyWithArray(ctx, 2);
RedisModule_ReplyWithString(ctx, keyname);
RedisModule_ReplyWithStringBuffer(ctx, data, len);
pd->nkeys++;
}
if (was_opened)
RedisModule_CloseKey(key);
}
int scan_strings(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
scan_strings_pd pd = {
.nkeys = 0,
};
RedisModule_ReplyWithArray(ctx, REDISMODULE_POSTPONED_ARRAY_LEN);
RedisModuleScanCursor* cursor = RedisModule_ScanCursorCreate();
while(RedisModule_Scan(ctx, cursor, scan_strings_callback, &pd));
RedisModule_ScanCursorDestroy(cursor);
RedisModule_ReplySetArrayLength(ctx, pd.nkeys);
return REDISMODULE_OK;
}
typedef struct {
RedisModuleCtx *ctx;
size_t nreplies;
} scan_key_pd;
void scan_key_callback(RedisModuleKey *key, RedisModuleString* field, RedisModuleString* value, void *privdata) {
REDISMODULE_NOT_USED(key);
scan_key_pd* pd = privdata;
RedisModule_ReplyWithArray(pd->ctx, 2);
RedisModule_ReplyWithString(pd->ctx, field);
if (value)
RedisModule_ReplyWithString(pd->ctx, value);
else
RedisModule_ReplyWithNull(pd->ctx);
pd->nreplies++;
}
int scan_key(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
{
if (argc != 2) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
scan_key_pd pd = {
.ctx = ctx,
.nreplies = 0,
};
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_READ);
if (!key) {
RedisModule_ReplyWithError(ctx, "not found");
return REDISMODULE_OK;
}
RedisModule_ReplyWithArray(ctx, REDISMODULE_POSTPONED_ARRAY_LEN);
RedisModuleScanCursor* cursor = RedisModule_ScanCursorCreate();
while(RedisModule_ScanKey(key, cursor, scan_key_callback, &pd));
RedisModule_ScanCursorDestroy(cursor);
RedisModule_ReplySetArrayLength(ctx, pd.nreplies);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
if (RedisModule_Init(ctx, "scan", 1, REDISMODULE_APIVER_1)== REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "scan.scan_strings", scan_strings, "", 0, 0, 0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "scan.scan_key", scan_key, "", 0, 0, 0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}
...@@ -15,11 +15,19 @@ RedisModuleString *after_str = NULL; ...@@ -15,11 +15,19 @@ RedisModuleString *after_str = NULL;
void *testrdb_type_load(RedisModuleIO *rdb, int encver) { void *testrdb_type_load(RedisModuleIO *rdb, int encver) {
int count = RedisModule_LoadSigned(rdb); int count = RedisModule_LoadSigned(rdb);
if (RedisModule_IsIOError(rdb)) RedisModuleString *str = RedisModule_LoadString(rdb);
float f = RedisModule_LoadFloat(rdb);
long double ld = RedisModule_LoadLongDouble(rdb);
if (RedisModule_IsIOError(rdb)) {
RedisModuleCtx *ctx = RedisModule_GetContextFromIO(rdb);
RedisModule_FreeString(ctx, str);
return NULL; return NULL;
}
/* Using the values only after checking for io errors. */
assert(count==1); assert(count==1);
assert(encver==1); assert(encver==1);
RedisModuleString *str = RedisModule_LoadString(rdb); assert(f==1.5f);
assert(ld==0.333333333333333333L);
return str; return str;
} }
...@@ -27,6 +35,8 @@ void testrdb_type_save(RedisModuleIO *rdb, void *value) { ...@@ -27,6 +35,8 @@ void testrdb_type_save(RedisModuleIO *rdb, void *value) {
RedisModuleString *str = (RedisModuleString*)value; RedisModuleString *str = (RedisModuleString*)value;
RedisModule_SaveSigned(rdb, 1); RedisModule_SaveSigned(rdb, 1);
RedisModule_SaveString(rdb, str); RedisModule_SaveString(rdb, str);
RedisModule_SaveFloat(rdb, 1.5);
RedisModule_SaveLongDouble(rdb, 0.333333333333333333L);
} }
void testrdb_aux_save(RedisModuleIO *rdb, int when) { void testrdb_aux_save(RedisModuleIO *rdb, int when) {
......
...@@ -11,28 +11,55 @@ proc fail {msg} { ...@@ -11,28 +11,55 @@ proc fail {msg} {
proc assert {condition} { proc assert {condition} {
if {![uplevel 1 [list expr $condition]]} { if {![uplevel 1 [list expr $condition]]} {
error "assertion:Expected condition '$condition' to be true ([uplevel 1 [list subst -nocommands $condition]])" set context "(context: [info frame -1])"
error "assertion:Expected [uplevel 1 [list subst -nocommands $condition]] $context"
} }
} }
proc assert_no_match {pattern value} { proc assert_no_match {pattern value} {
if {[string match $pattern $value]} { if {[string match $pattern $value]} {
error "assertion:Expected '$value' to not match '$pattern'" set context "(context: [info frame -1])"
error "assertion:Expected '$value' to not match '$pattern' $context"
} }
} }
proc assert_match {pattern value} { proc assert_match {pattern value} {
if {![string match $pattern $value]} { if {![string match $pattern $value]} {
error "assertion:Expected '$value' to match '$pattern'" set context "(context: [info frame -1])"
error "assertion:Expected '$value' to match '$pattern' $context"
} }
} }
proc assert_equal {expected value {detail ""}} { proc assert_equal {value expected {detail ""}} {
if {$expected ne $value} { if {$expected ne $value} {
if {$detail ne ""} { if {$detail ne ""} {
set detail " (detail: $detail)" set detail "(detail: $detail)"
} else {
set detail "(context: [info frame -1])"
}
error "assertion:Expected '$value' to be equal to '$expected' $detail"
}
}
proc assert_lessthan {value expected {detail ""}} {
if {!($value < $expected)} {
if {$detail ne ""} {
set detail "(detail: $detail)"
} else {
set detail "(context: [info frame -1])"
}
error "assertion:Expected '$value' to be lessthan to '$expected' $detail"
}
}
proc assert_range {value min max {detail ""}} {
if {!($value <= $max && $value >= $min)} {
if {$detail ne ""} {
set detail "(detail: $detail)"
} else {
set detail "(context: [info frame -1])"
} }
error "assertion:Expected '$value' to be equal to '$expected'$detail" error "assertion:Expected '$value' to be between to '$min' and '$max' $detail"
} }
} }
......
set testmodule [file normalize tests/modules/blockonkeys.so]
start_server {tags {"modules"}} {
r module load $testmodule
test {Module client blocked on keys (no metadata): No block} {
r del k
r fsl.push k 33
r fsl.push k 34
r fsl.bpop2 k 0
} {34 33}
test {Module client blocked on keys (no metadata): Timeout} {
r del k
set rd [redis_deferring_client]
r fsl.push k 33
$rd fsl.bpop2 k 1
assert_equal {Request timedout} [$rd read]
}
test {Module client blocked on keys (no metadata): Blocked, case 1} {
r del k
set rd [redis_deferring_client]
r fsl.push k 33
$rd fsl.bpop2 k 0
r fsl.push k 34
assert_equal {34 33} [$rd read]
}
test {Module client blocked on keys (no metadata): Blocked, case 2} {
r del k
set rd [redis_deferring_client]
r fsl.push k 33
r fsl.push k 34
$rd fsl.bpop2 k 0
assert_equal {34 33} [$rd read]
}
test {Module client blocked on keys (with metadata): No block} {
r del k
r fsl.push k 34
r fsl.bpopgt k 30 0
} {34}
test {Module client blocked on keys (with metadata): Timeout} {
r del k
set rd [redis_deferring_client]
r fsl.push k 33
$rd fsl.bpopgt k 35 1
assert_equal {Request timedout} [$rd read]
}
test {Module client blocked on keys (with metadata): Blocked, case 1} {
r del k
set rd [redis_deferring_client]
r fsl.push k 33
$rd fsl.bpopgt k 33 0
r fsl.push k 34
assert_equal {34} [$rd read]
}
test {Module client blocked on keys (with metadata): Blocked, case 2} {
r del k
set rd [redis_deferring_client]
$rd fsl.bpopgt k 35 0
r fsl.push k 33
r fsl.push k 34
r fsl.push k 35
r fsl.push k 36
assert_equal {36} [$rd read]
}
test {Module client blocked on keys does not wake up on wrong type} {
r del k
set rd [redis_deferring_client]
$rd fsl.bpop2 k 0
r lpush k 12
r lpush k 13
r lpush k 14
r del k
r fsl.push k 33
r fsl.push k 34
assert_equal {34 33} [$rd read]
}
}
set testmodule [file normalize tests/modules/datatype.so]
start_server {tags {"modules"}} {
r module load $testmodule
test {DataType: Test module is sane, GET/SET work.} {
r datatype.set dtkey 100 stringval
assert {[r datatype.get dtkey] eq {100 stringval}}
}
test {DataType: RM_SaveDataTypeToString(), RM_LoadDataTypeFromString() work} {
r datatype.set dtkey -1111 MyString
set encoded [r datatype.dump dtkey]
r datatype.restore dtkeycopy $encoded
assert {[r datatype.get dtkeycopy] eq {-1111 MyString}}
}
test {DataType: Handle truncated RM_LoadDataTypeFromString()} {
r datatype.set dtkey -1111 MyString
set encoded [r datatype.dump dtkey]
set truncated [string range $encoded 0 end-1]
catch {r datatype.restore dtkeycopy $truncated} e
set e
} {*Invalid*}
}
...@@ -3,26 +3,138 @@ set testmodule [file normalize tests/modules/hooks.so] ...@@ -3,26 +3,138 @@ set testmodule [file normalize tests/modules/hooks.so]
tags "modules" { tags "modules" {
start_server {} { start_server {} {
r module load $testmodule r module load $testmodule
r config set appendonly yes
test {Test clients connection / disconnection hooks} { test {Test clients connection / disconnection hooks} {
for {set j 0} {$j < 2} {incr j} { for {set j 0} {$j < 2} {incr j} {
set rd1 [redis_deferring_client] set rd1 [redis_deferring_client]
$rd1 close $rd1 close
} }
assert {[r llen connected] > 1} assert {[r hooks.event_count client-connected] > 1}
assert {[r llen disconnected] > 1} assert {[r hooks.event_count client-disconnected] > 1}
}
test {Test module cron hook} {
after 100
assert {[r hooks.event_count cron-loop] > 0}
set hz [r hooks.event_last cron-loop]
assert_equal $hz 10
}
test {Test module loaded / unloaded hooks} {
set othermodule [file normalize tests/modules/infotest.so]
r module load $othermodule
r module unload infotest
assert_equal [r hooks.event_last module-loaded] "infotest"
assert_equal [r hooks.event_last module-unloaded] "infotest"
}
test {Test module aofrw hook} {
r debug populate 1000 foo 10000 ;# 10mb worth of data
r config set rdbcompression no ;# rdb progress is only checked once in 2mb
r BGREWRITEAOF
waitForBgrewriteaof r
assert_equal [string match {*module-event-persistence-aof-start*} [exec tail -20 < [srv 0 stdout]]] 1
assert_equal [string match {*module-event-persistence-end*} [exec tail -20 < [srv 0 stdout]]] 1
}
test {Test module aof load and rdb/aof progress hooks} {
# create some aof tail (progress is checked only once in 1000 commands)
for {set j 0} {$j < 4000} {incr j} {
r set "bar$j" x
}
# set some configs that will cause many loading progress events during aof loading
r config set key-load-delay 1
r config set dynamic-hz no
r config set hz 500
r DEBUG LOADAOF
assert_equal [r hooks.event_last loading-aof-start] 0
assert_equal [r hooks.event_last loading-end] 0
assert {[r hooks.event_count loading-rdb-start] == 0}
assert {[r hooks.event_count loading-progress-rdb] >= 2} ;# comes from the preamble section
assert {[r hooks.event_count loading-progress-aof] >= 2}
}
# undo configs before next test
r config set dynamic-hz yes
r config set key-load-delay 0
test {Test module rdb save hook} {
# debug reload does: save, flush, load:
assert {[r hooks.event_count persistence-syncrdb-start] == 0}
assert {[r hooks.event_count loading-rdb-start] == 0}
r debug reload
assert {[r hooks.event_count persistence-syncrdb-start] == 1}
assert {[r hooks.event_count loading-rdb-start] == 1}
} }
test {Test flushdb hooks} { test {Test flushdb hooks} {
r flushall ;# Note: only the "end" RPUSH will survive
r select 1
r flushdb r flushdb
r select 2 assert_equal [r hooks.event_last flush-start] 9
r flushdb assert_equal [r hooks.event_last flush-end] 9
r select 9 r flushall
assert {[r llen flush-start] == 2} assert_equal [r hooks.event_last flush-start] -1
assert {[r llen flush-end] == 3} assert_equal [r hooks.event_last flush-end] -1
assert {[r lrange flush-start 0 -1] eq {1 2}} }
assert {[r lrange flush-end 0 -1] eq {-1 1 2}}
# replication related tests
set master [srv 0 client]
set master_host [srv 0 host]
set master_port [srv 0 port]
start_server {} {
r module load $testmodule
set replica [srv 0 client]
set replica_host [srv 0 host]
set replica_port [srv 0 port]
$replica replicaof $master_host $master_port
wait_for_condition 50 100 {
[string match {*master_link_status:up*} [r info replication]]
} else {
fail "Can't turn the instance into a replica"
} }
test {Test master link up hook} {
assert_equal [r hooks.event_count masterlink-up] 1
assert_equal [r hooks.event_count masterlink-down] 0
}
test {Test role-replica hook} {
assert_equal [r hooks.event_count role-replica] 1
assert_equal [r hooks.event_count role-master] 0
assert_equal [r hooks.event_last role-replica] [s 0 master_host]
}
test {Test replica-online hook} {
assert_equal [r -1 hooks.event_count replica-online] 1
assert_equal [r -1 hooks.event_count replica-offline] 0
}
test {Test master link down hook} {
r client kill type master
assert_equal [r hooks.event_count masterlink-down] 1
}
$replica replicaof no one
test {Test role-master hook} {
assert_equal [r hooks.event_count role-replica] 1
assert_equal [r hooks.event_count role-master] 1
assert_equal [r hooks.event_last role-master] {}
}
test {Test replica-offline hook} {
assert_equal [r -1 hooks.event_count replica-online] 1
assert_equal [r -1 hooks.event_count replica-offline] 1
}
# get the replica stdout, to be used by the next test
set replica_stdout [srv 0 stdout]
}
# look into the log file of the server that just exited
test {Test shutdown hook} {
assert_equal [string match {*module-event-shutdown*} [exec tail -5 < $replica_stdout]] 1
}
} }
} }
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