Unverified Commit 18b62a51 authored by Salvatore Sanfilippo's avatar Salvatore Sanfilippo Committed by GitHub
Browse files

Merge branch 'unstable' into module_lru_lfu

parents e978bdf9 327cb45e
...@@ -21,4 +21,5 @@ $TCLSH tests/test_helper.tcl \ ...@@ -21,4 +21,5 @@ $TCLSH tests/test_helper.tcl \
--single unit/moduleapi/propagate \ --single unit/moduleapi/propagate \
--single unit/moduleapi/hooks \ --single unit/moduleapi/hooks \
--single unit/moduleapi/misc \ --single unit/moduleapi/misc \
--single unit/moduleapi/blockonkeys \
"${@}" "${@}"
...@@ -731,7 +731,7 @@ int loadAppendOnlyFile(char *filename) { ...@@ -731,7 +731,7 @@ int loadAppendOnlyFile(char *filename) {
server.aof_state = AOF_OFF; server.aof_state = AOF_OFF;
fakeClient = createAOFClient(); fakeClient = createAOFClient();
startLoadingFile(fp, filename); startLoadingFile(fp, filename, RDBFLAGS_AOF_PREAMBLE);
/* Check if this AOF file has an RDB preamble. In that case we need to /* Check if this AOF file has an RDB preamble. In that case we need to
* load the RDB file and later continue loading the AOF tail. */ * load the RDB file and later continue loading the AOF tail. */
...@@ -746,7 +746,7 @@ int loadAppendOnlyFile(char *filename) { ...@@ -746,7 +746,7 @@ int loadAppendOnlyFile(char *filename) {
serverLog(LL_NOTICE,"Reading RDB preamble from AOF file..."); serverLog(LL_NOTICE,"Reading RDB preamble from AOF file...");
if (fseek(fp,0,SEEK_SET) == -1) goto readerr; if (fseek(fp,0,SEEK_SET) == -1) goto readerr;
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
if (rdbLoadRio(&rdb,NULL,1) != C_OK) { if (rdbLoadRio(&rdb,RDBFLAGS_AOF_PREAMBLE,NULL) != C_OK) {
serverLog(LL_WARNING,"Error reading the RDB preamble of the AOF file, AOF loading aborted"); serverLog(LL_WARNING,"Error reading the RDB preamble of the AOF file, AOF loading aborted");
goto readerr; goto readerr;
} else { } else {
...@@ -767,6 +767,7 @@ int loadAppendOnlyFile(char *filename) { ...@@ -767,6 +767,7 @@ int loadAppendOnlyFile(char *filename) {
if (!(loops++ % 1000)) { if (!(loops++ % 1000)) {
loadingProgress(ftello(fp)); loadingProgress(ftello(fp));
processEventsWhileBlocked(); processEventsWhileBlocked();
processModuleLoadingProgressEvent(1);
} }
if (fgets(buf,sizeof(buf),fp) == NULL) { if (fgets(buf,sizeof(buf),fp) == NULL) {
...@@ -859,7 +860,7 @@ loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */ ...@@ -859,7 +860,7 @@ loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */
fclose(fp); fclose(fp);
freeFakeClient(fakeClient); freeFakeClient(fakeClient);
server.aof_state = old_aof_state; server.aof_state = old_aof_state;
stopLoading(); stopLoading(1);
aofUpdateCurrentSize(); aofUpdateCurrentSize();
server.aof_rewrite_base_size = server.aof_current_size; server.aof_rewrite_base_size = server.aof_current_size;
server.aof_fsync_offset = server.aof_current_size; server.aof_fsync_offset = server.aof_current_size;
...@@ -1400,9 +1401,11 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -1400,9 +1401,11 @@ int rewriteAppendOnlyFile(char *filename) {
if (server.aof_rewrite_incremental_fsync) if (server.aof_rewrite_incremental_fsync)
rioSetAutoSync(&aof,REDIS_AUTOSYNC_BYTES); rioSetAutoSync(&aof,REDIS_AUTOSYNC_BYTES);
startSaving(RDBFLAGS_AOF_PREAMBLE);
if (server.aof_use_rdb_preamble) { if (server.aof_use_rdb_preamble) {
int error; int error;
if (rdbSaveRio(&aof,&error,RDB_SAVE_AOF_PREAMBLE,NULL) == C_ERR) { if (rdbSaveRio(&aof,&error,RDBFLAGS_AOF_PREAMBLE,NULL) == C_ERR) {
errno = error; errno = error;
goto werr; goto werr;
} }
...@@ -1465,15 +1468,18 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -1465,15 +1468,18 @@ int rewriteAppendOnlyFile(char *filename) {
if (rename(tmpfile,filename) == -1) { if (rename(tmpfile,filename) == -1) {
serverLog(LL_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno)); serverLog(LL_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
unlink(tmpfile); unlink(tmpfile);
stopSaving(0);
return C_ERR; return C_ERR;
} }
serverLog(LL_NOTICE,"SYNC append only file rewrite performed"); serverLog(LL_NOTICE,"SYNC append only file rewrite performed");
stopSaving(1);
return C_OK; return C_OK;
werr: werr:
serverLog(LL_WARNING,"Write error writing append only file on disk: %s", strerror(errno)); serverLog(LL_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
fclose(fp); fclose(fp);
unlink(tmpfile); unlink(tmpfile);
stopSaving(0);
return C_ERR; return C_ERR;
} }
......
...@@ -174,6 +174,7 @@ void unblockClient(client *c) { ...@@ -174,6 +174,7 @@ void unblockClient(client *c) {
} else if (c->btype == BLOCKED_WAIT) { } else if (c->btype == BLOCKED_WAIT) {
unblockClientWaitingReplicas(c); unblockClientWaitingReplicas(c);
} else if (c->btype == BLOCKED_MODULE) { } else if (c->btype == BLOCKED_MODULE) {
if (moduleClientIsBlockedOnKeys(c)) unblockClientWaitingData(c);
unblockClientFromModule(c); unblockClientFromModule(c);
} else { } else {
serverPanic("Unknown btype in unblockClient()."); serverPanic("Unknown btype in unblockClient().");
...@@ -430,6 +431,49 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) { ...@@ -430,6 +431,49 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
} }
} }
/* Helper function for handleClientsBlockedOnKeys(). This function is called
* in order to check if we can serve clients blocked by modules using
* RM_BlockClientOnKeys(), when the corresponding key was signaled as ready:
* our goal here is to call the RedisModuleBlockedClient reply() callback to
* see if the key is really able to serve the client, and in that case,
* unblock it. */
void serveClientsBlockedOnKeyByModule(readyList *rl) {
dictEntry *de;
/* We serve clients in the same order they blocked for
* this key, from the first blocked to the last. */
de = dictFind(rl->db->blocking_keys,rl->key);
if (de) {
list *clients = dictGetVal(de);
int numclients = listLength(clients);
while(numclients--) {
listNode *clientnode = listFirst(clients);
client *receiver = clientnode->value;
/* Put at the tail, so that at the next call
* we'll not run into it again: clients here may not be
* ready to be served, so they'll remain in the list
* sometimes. We want also be able to skip clients that are
* not blocked for the MODULE type safely. */
listDelNode(clients,clientnode);
listAddNodeTail(clients,receiver);
if (receiver->btype != BLOCKED_MODULE) continue;
/* Note that if *this* client cannot be served by this key,
* it does not mean that another client that is next into the
* list cannot be served as well: they may be blocked by
* different modules with different triggers to consider if a key
* is ready or not. This means we can't exit the loop but need
* to continue after the first failure. */
if (!moduleTryServeClientBlockedOnKey(receiver, rl->key)) continue;
moduleUnblockClient(receiver);
}
}
}
/* This function should be called by Redis every time a single command, /* This function should be called by Redis every time a single command,
* a MULTI/EXEC block, or a Lua script, terminated its execution after * a MULTI/EXEC block, or a Lua script, terminated its execution after
* being called by a client. It handles serving clients blocked in * being called by a client. It handles serving clients blocked in
...@@ -480,6 +524,10 @@ void handleClientsBlockedOnKeys(void) { ...@@ -480,6 +524,10 @@ void handleClientsBlockedOnKeys(void) {
serveClientsBlockedOnSortedSetKey(o,rl); serveClientsBlockedOnSortedSetKey(o,rl);
else if (o->type == OBJ_STREAM) else if (o->type == OBJ_STREAM)
serveClientsBlockedOnStreamKey(o,rl); serveClientsBlockedOnStreamKey(o,rl);
/* We want to serve clients blocked on module keys
* regardless of the object type: we don't know what the
* module is trying to accomplish right now. */
serveClientsBlockedOnKeyByModule(rl);
} }
/* Free this item. */ /* Free this item. */
......
...@@ -465,6 +465,29 @@ int getFlushCommandFlags(client *c, int *flags) { ...@@ -465,6 +465,29 @@ int getFlushCommandFlags(client *c, int *flags) {
return C_OK; return C_OK;
} }
/* Flushes the whole server data set. */
void flushAllDataAndResetRDB(int flags) {
server.dirty += emptyDb(-1,flags,NULL);
if (server.rdb_child_pid != -1) killRDBChild();
if (server.saveparamslen > 0) {
/* Normally rdbSave() will reset dirty, but we don't want this here
* as otherwise FLUSHALL will not be replicated nor put into the AOF. */
int saved_dirty = server.dirty;
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
rdbSave(server.rdb_filename,rsiptr);
server.dirty = saved_dirty;
}
server.dirty++;
#if defined(USE_JEMALLOC)
/* jemalloc 5 doesn't release pages back to the OS when there's no traffic.
* for large databases, flushdb blocks for long anyway, so a bit more won't
* harm and this way the flush and purge will be synchroneus. */
if (!(flags & EMPTYDB_ASYNC))
jemalloc_purge();
#endif
}
/* FLUSHDB [ASYNC] /* FLUSHDB [ASYNC]
* *
* Flushes the currently SELECTed Redis DB. */ * Flushes the currently SELECTed Redis DB. */
...@@ -488,28 +511,9 @@ void flushdbCommand(client *c) { ...@@ -488,28 +511,9 @@ void flushdbCommand(client *c) {
* Flushes the whole server data set. */ * Flushes the whole server data set. */
void flushallCommand(client *c) { void flushallCommand(client *c) {
int flags; int flags;
if (getFlushCommandFlags(c,&flags) == C_ERR) return; if (getFlushCommandFlags(c,&flags) == C_ERR) return;
server.dirty += emptyDb(-1,flags,NULL); flushAllDataAndResetRDB(flags);
addReply(c,shared.ok); addReply(c,shared.ok);
if (server.rdb_child_pid != -1) killRDBChild();
if (server.saveparamslen > 0) {
/* Normally rdbSave() will reset dirty, but we don't want this here
* as otherwise FLUSHALL will not be replicated nor put into the AOF. */
int saved_dirty = server.dirty;
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
rdbSave(server.rdb_filename,rsiptr);
server.dirty = saved_dirty;
}
server.dirty++;
#if defined(USE_JEMALLOC)
/* jemalloc 5 doesn't release pages back to the OS when there's no traffic.
* for large databases, flushdb blocks for long anyway, so a bit more won't
* harm and this way the flush and purge will be synchroneus. */
if (!(flags & EMPTYDB_ASYNC))
jemalloc_purge();
#endif
} }
/* This command implements DEL and LAZYDEL. */ /* This command implements DEL and LAZYDEL. */
......
...@@ -417,7 +417,7 @@ NULL ...@@ -417,7 +417,7 @@ NULL
} }
emptyDb(-1,EMPTYDB_NO_FLAGS,NULL); emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
protectClient(c); protectClient(c);
int ret = rdbLoad(server.rdb_filename,NULL); int ret = rdbLoad(server.rdb_filename,NULL,RDBFLAGS_NONE);
unprotectClient(c); unprotectClient(c);
if (ret != C_OK) { if (ret != C_OK) {
addReplyError(c,"Error trying to load the RDB dump"); addReplyError(c,"Error trying to load the RDB dump");
......
This diff is collapsed.
...@@ -129,6 +129,7 @@ int HelloTypeInsert_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, ...@@ -129,6 +129,7 @@ int HelloTypeInsert_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv,
/* Insert the new element. */ /* Insert the new element. */
HelloTypeInsert(hto,value); HelloTypeInsert(hto,value);
RedisModule_SignalKeyAsReady(ctx,argv[1]);
RedisModule_ReplyWithLongLong(ctx,hto->len); RedisModule_ReplyWithLongLong(ctx,hto->len);
RedisModule_ReplicateVerbatim(ctx); RedisModule_ReplicateVerbatim(ctx);
...@@ -190,6 +191,77 @@ int HelloTypeLen_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int ...@@ -190,6 +191,77 @@ int HelloTypeLen_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int
return REDISMODULE_OK; return REDISMODULE_OK;
} }
/* ====================== Example of a blocking command ==================== */
/* Reply callback for blocking command HELLOTYPE.BRANGE, this will get
* called when the key we blocked for is ready: we need to check if we
* can really serve the client, and reply OK or ERR accordingly. */
int HelloBlock_Reply(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
RedisModuleString *keyname = RedisModule_GetBlockedClientReadyKey(ctx);
RedisModuleKey *key = RedisModule_OpenKey(ctx,keyname,REDISMODULE_READ);
int type = RedisModule_KeyType(key);
if (type != REDISMODULE_KEYTYPE_MODULE ||
RedisModule_ModuleTypeGetType(key) != HelloType)
{
RedisModule_CloseKey(key);
return REDISMODULE_ERR;
}
/* In case the key is able to serve our blocked client, let's directly
* use our original command implementation to make this example simpler. */
RedisModule_CloseKey(key);
return HelloTypeRange_RedisCommand(ctx,argv,argc-1);
}
/* Timeout callback for blocking command HELLOTYPE.BRANGE */
int HelloBlock_Timeout(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
return RedisModule_ReplyWithSimpleString(ctx,"Request timedout");
}
/* Private data freeing callback for HELLOTYPE.BRANGE command. */
void HelloBlock_FreeData(RedisModuleCtx *ctx, void *privdata) {
REDISMODULE_NOT_USED(ctx);
RedisModule_Free(privdata);
}
/* HELLOTYPE.BRANGE key first count timeout -- This is a blocking verison of
* the RANGE operation, in order to show how to use the API
* RedisModule_BlockClientOnKeys(). */
int HelloTypeBRange_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 5) return RedisModule_WrongArity(ctx);
RedisModule_AutoMemory(ctx); /* Use automatic memory management. */
RedisModuleKey *key = RedisModule_OpenKey(ctx,argv[1],
REDISMODULE_READ|REDISMODULE_WRITE);
int type = RedisModule_KeyType(key);
if (type != REDISMODULE_KEYTYPE_EMPTY &&
RedisModule_ModuleTypeGetType(key) != HelloType)
{
return RedisModule_ReplyWithError(ctx,REDISMODULE_ERRORMSG_WRONGTYPE);
}
/* Parse the timeout before even trying to serve the client synchronously,
* so that we always fail ASAP on syntax errors. */
long long timeout;
if (RedisModule_StringToLongLong(argv[4],&timeout) != REDISMODULE_OK) {
return RedisModule_ReplyWithError(ctx,
"ERR invalid timeout parameter");
}
/* Can we serve the reply synchronously? */
if (type != REDISMODULE_KEYTYPE_EMPTY) {
return HelloTypeRange_RedisCommand(ctx,argv,argc-1);
}
/* Otherwise let's block on the key. */
void *privdata = RedisModule_Alloc(100);
RedisModule_BlockClientOnKeys(ctx,HelloBlock_Reply,HelloBlock_Timeout,HelloBlock_FreeData,timeout,argv+1,1,privdata);
return REDISMODULE_OK;
}
/* ========================== "hellotype" type methods ======================= */ /* ========================== "hellotype" type methods ======================= */
...@@ -282,5 +354,9 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -282,5 +354,9 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
HelloTypeLen_RedisCommand,"readonly",1,1,1) == REDISMODULE_ERR) HelloTypeLen_RedisCommand,"readonly",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"hellotype.brange",
HelloTypeBRange_RedisCommand,"readonly",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK; return REDISMODULE_OK;
} }
...@@ -1118,6 +1118,11 @@ void freeClient(client *c) { ...@@ -1118,6 +1118,11 @@ void freeClient(client *c) {
if (c->flags & CLIENT_SLAVE && listLength(server.slaves) == 0) if (c->flags & CLIENT_SLAVE && listLength(server.slaves) == 0)
server.repl_no_slaves_since = server.unixtime; server.repl_no_slaves_since = server.unixtime;
refreshGoodSlavesCount(); refreshGoodSlavesCount();
/* Fire the replica change modules event. */
if (c->replstate == SLAVE_STATE_ONLINE)
moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE,
REDISMODULE_SUBEVENT_REPLICA_CHANGE_OFFLINE,
NULL);
} }
/* Master/slave cleanup Case 2: /* Master/slave cleanup Case 2:
......
...@@ -1080,9 +1080,9 @@ ssize_t rdbSaveAuxFieldStrInt(rio *rdb, char *key, long long val) { ...@@ -1080,9 +1080,9 @@ ssize_t rdbSaveAuxFieldStrInt(rio *rdb, char *key, long long val) {
} }
/* Save a few default AUX fields with information about the RDB generated. */ /* Save a few default AUX fields with information about the RDB generated. */
int rdbSaveInfoAuxFields(rio *rdb, int flags, rdbSaveInfo *rsi) { int rdbSaveInfoAuxFields(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
int redis_bits = (sizeof(void*) == 8) ? 64 : 32; int redis_bits = (sizeof(void*) == 8) ? 64 : 32;
int aof_preamble = (flags & RDB_SAVE_AOF_PREAMBLE) != 0; int aof_preamble = (rdbflags & RDBFLAGS_AOF_PREAMBLE) != 0;
/* Add a few fields about the state when the RDB was created. */ /* Add a few fields about the state when the RDB was created. */
if (rdbSaveAuxFieldStrStr(rdb,"redis-ver",REDIS_VERSION) == -1) return -1; if (rdbSaveAuxFieldStrStr(rdb,"redis-ver",REDIS_VERSION) == -1) return -1;
...@@ -1150,7 +1150,7 @@ ssize_t rdbSaveSingleModuleAux(rio *rdb, int when, moduleType *mt) { ...@@ -1150,7 +1150,7 @@ ssize_t rdbSaveSingleModuleAux(rio *rdb, int when, moduleType *mt) {
* When the function returns C_ERR and if 'error' is not NULL, the * When the function returns C_ERR and if 'error' is not NULL, the
* integer pointed by 'error' is set to the value of errno just after the I/O * integer pointed by 'error' is set to the value of errno just after the I/O
* error. */ * error. */
int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) { int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) {
dictIterator *di = NULL; dictIterator *di = NULL;
dictEntry *de; dictEntry *de;
char magic[10]; char magic[10];
...@@ -1162,7 +1162,7 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) { ...@@ -1162,7 +1162,7 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
rdb->update_cksum = rioGenericUpdateChecksum; rdb->update_cksum = rioGenericUpdateChecksum;
snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION); snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION);
if (rdbWriteRaw(rdb,magic,9) == -1) goto werr; if (rdbWriteRaw(rdb,magic,9) == -1) goto werr;
if (rdbSaveInfoAuxFields(rdb,flags,rsi) == -1) goto werr; if (rdbSaveInfoAuxFields(rdb,rdbflags,rsi) == -1) goto werr;
if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_BEFORE_RDB) == -1) goto werr; if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_BEFORE_RDB) == -1) goto werr;
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
...@@ -1199,7 +1199,7 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) { ...@@ -1199,7 +1199,7 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
/* When this RDB is produced as part of an AOF rewrite, move /* When this RDB is produced as part of an AOF rewrite, move
* accumulated diff from parent to child while rewriting in * accumulated diff from parent to child while rewriting in
* order to have a smaller final write. */ * order to have a smaller final write. */
if (flags & RDB_SAVE_AOF_PREAMBLE && if (rdbflags & RDBFLAGS_AOF_PREAMBLE &&
rdb->processed_bytes > processed+AOF_READ_DIFF_INTERVAL_BYTES) rdb->processed_bytes > processed+AOF_READ_DIFF_INTERVAL_BYTES)
{ {
processed = rdb->processed_bytes; processed = rdb->processed_bytes;
...@@ -1254,18 +1254,21 @@ werr: ...@@ -1254,18 +1254,21 @@ werr:
int rdbSaveRioWithEOFMark(rio *rdb, int *error, rdbSaveInfo *rsi) { int rdbSaveRioWithEOFMark(rio *rdb, int *error, rdbSaveInfo *rsi) {
char eofmark[RDB_EOF_MARK_SIZE]; char eofmark[RDB_EOF_MARK_SIZE];
startSaving(RDBFLAGS_REPLICATION);
getRandomHexChars(eofmark,RDB_EOF_MARK_SIZE); getRandomHexChars(eofmark,RDB_EOF_MARK_SIZE);
if (error) *error = 0; if (error) *error = 0;
if (rioWrite(rdb,"$EOF:",5) == 0) goto werr; if (rioWrite(rdb,"$EOF:",5) == 0) goto werr;
if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr; if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr;
if (rioWrite(rdb,"\r\n",2) == 0) goto werr; if (rioWrite(rdb,"\r\n",2) == 0) goto werr;
if (rdbSaveRio(rdb,error,RDB_SAVE_NONE,rsi) == C_ERR) goto werr; if (rdbSaveRio(rdb,error,RDBFLAGS_NONE,rsi) == C_ERR) goto werr;
if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr; if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr;
stopSaving(1);
return C_OK; return C_OK;
werr: /* Write error. */ werr: /* Write error. */
/* Set 'error' only if not already set by rdbSaveRio() call. */ /* Set 'error' only if not already set by rdbSaveRio() call. */
if (error && *error == 0) *error = errno; if (error && *error == 0) *error = errno;
stopSaving(0);
return C_ERR; return C_ERR;
} }
...@@ -1291,11 +1294,12 @@ int rdbSave(char *filename, rdbSaveInfo *rsi) { ...@@ -1291,11 +1294,12 @@ int rdbSave(char *filename, rdbSaveInfo *rsi) {
} }
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
startSaving(RDBFLAGS_NONE);
if (server.rdb_save_incremental_fsync) if (server.rdb_save_incremental_fsync)
rioSetAutoSync(&rdb,REDIS_AUTOSYNC_BYTES); rioSetAutoSync(&rdb,REDIS_AUTOSYNC_BYTES);
if (rdbSaveRio(&rdb,&error,RDB_SAVE_NONE,rsi) == C_ERR) { if (rdbSaveRio(&rdb,&error,RDBFLAGS_NONE,rsi) == C_ERR) {
errno = error; errno = error;
goto werr; goto werr;
} }
...@@ -1317,6 +1321,7 @@ int rdbSave(char *filename, rdbSaveInfo *rsi) { ...@@ -1317,6 +1321,7 @@ int rdbSave(char *filename, rdbSaveInfo *rsi) {
cwdp ? cwdp : "unknown", cwdp ? cwdp : "unknown",
strerror(errno)); strerror(errno));
unlink(tmpfile); unlink(tmpfile);
stopSaving(0);
return C_ERR; return C_ERR;
} }
...@@ -1324,12 +1329,14 @@ int rdbSave(char *filename, rdbSaveInfo *rsi) { ...@@ -1324,12 +1329,14 @@ int rdbSave(char *filename, rdbSaveInfo *rsi) {
server.dirty = 0; server.dirty = 0;
server.lastsave = time(NULL); server.lastsave = time(NULL);
server.lastbgsave_status = C_OK; server.lastbgsave_status = C_OK;
stopSaving(1);
return C_OK; return C_OK;
werr: werr:
serverLog(LL_WARNING,"Write error saving DB on disk: %s", strerror(errno)); serverLog(LL_WARNING,"Write error saving DB on disk: %s", strerror(errno));
fclose(fp); fclose(fp);
unlink(tmpfile); unlink(tmpfile);
stopSaving(0);
return C_ERR; return C_ERR;
} }
...@@ -1918,23 +1925,33 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1918,23 +1925,33 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
/* Mark that we are loading in the global state and setup the fields /* Mark that we are loading in the global state and setup the fields
* needed to provide loading stats. */ * needed to provide loading stats. */
void startLoading(size_t size) { void startLoading(size_t size, int rdbflags) {
/* Load the DB */ /* Load the DB */
server.loading = 1; server.loading = 1;
server.loading_start_time = time(NULL); server.loading_start_time = time(NULL);
server.loading_loaded_bytes = 0; server.loading_loaded_bytes = 0;
server.loading_total_bytes = size; server.loading_total_bytes = size;
/* Fire the loading modules start event. */
int subevent;
if (rdbflags & RDBFLAGS_AOF_PREAMBLE)
subevent = REDISMODULE_SUBEVENT_LOADING_AOF_START;
else if(rdbflags & RDBFLAGS_REPLICATION)
subevent = REDISMODULE_SUBEVENT_LOADING_REPL_START;
else
subevent = REDISMODULE_SUBEVENT_LOADING_RDB_START;
moduleFireServerEvent(REDISMODULE_EVENT_LOADING,subevent,NULL);
} }
/* Mark that we are loading in the global state and setup the fields /* Mark that we are loading in the global state and setup the fields
* needed to provide loading stats. * needed to provide loading stats.
* 'filename' is optional and used for rdb-check on error */ * 'filename' is optional and used for rdb-check on error */
void startLoadingFile(FILE *fp, char* filename) { void startLoadingFile(FILE *fp, char* filename, int rdbflags) {
struct stat sb; struct stat sb;
if (fstat(fileno(fp), &sb) == -1) if (fstat(fileno(fp), &sb) == -1)
sb.st_size = 0; sb.st_size = 0;
rdbFileBeingLoaded = filename; rdbFileBeingLoaded = filename;
startLoading(sb.st_size); startLoading(sb.st_size, rdbflags);
} }
/* Refresh the loading progress info */ /* Refresh the loading progress info */
...@@ -1945,9 +1962,37 @@ void loadingProgress(off_t pos) { ...@@ -1945,9 +1962,37 @@ void loadingProgress(off_t pos) {
} }
/* Loading finished */ /* Loading finished */
void stopLoading(void) { void stopLoading(int success) {
server.loading = 0; server.loading = 0;
rdbFileBeingLoaded = NULL; rdbFileBeingLoaded = NULL;
/* Fire the loading modules end event. */
moduleFireServerEvent(REDISMODULE_EVENT_LOADING,
success?
REDISMODULE_SUBEVENT_LOADING_ENDED:
REDISMODULE_SUBEVENT_LOADING_FAILED,
NULL);
}
void startSaving(int rdbflags) {
/* Fire the persistence modules end event. */
int subevent;
if (rdbflags & RDBFLAGS_AOF_PREAMBLE)
subevent = REDISMODULE_SUBEVENT_PERSISTENCE_AOF_START;
else if (getpid()!=server.pid)
subevent = REDISMODULE_SUBEVENT_PERSISTENCE_RDB_START;
else
subevent = REDISMODULE_SUBEVENT_PERSISTENCE_SYNC_RDB_START;
moduleFireServerEvent(REDISMODULE_EVENT_PERSISTENCE,subevent,NULL);
}
void stopSaving(int success) {
/* Fire the persistence modules end event. */
moduleFireServerEvent(REDISMODULE_EVENT_PERSISTENCE,
success?
REDISMODULE_SUBEVENT_PERSISTENCE_ENDED:
REDISMODULE_SUBEVENT_PERSISTENCE_FAILED,
NULL);
} }
/* Track loading progress in order to serve client's from time to time /* Track loading progress in order to serve client's from time to time
...@@ -1966,12 +2011,13 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) { ...@@ -1966,12 +2011,13 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
replicationSendNewlineToMaster(); replicationSendNewlineToMaster();
loadingProgress(r->processed_bytes); loadingProgress(r->processed_bytes);
processEventsWhileBlocked(); processEventsWhileBlocked();
processModuleLoadingProgressEvent(0);
} }
} }
/* Load an RDB file from the rio stream 'rdb'. On success C_OK is returned, /* Load an RDB file from the rio stream 'rdb'. On success C_OK is returned,
* otherwise C_ERR is returned and 'errno' is set accordingly. */ * otherwise C_ERR is returned and 'errno' is set accordingly. */
int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) { int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
uint64_t dbid; uint64_t dbid;
int type, rdbver; int type, rdbver;
redisDb *db = server.db+0; redisDb *db = server.db+0;
...@@ -2182,7 +2228,7 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) { ...@@ -2182,7 +2228,7 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
* received from the master. In the latter case, the master is * received from the master. In the latter case, the master is
* responsible for key expiry. If we would expire keys here, the * responsible for key expiry. If we would expire keys here, the
* snapshot taken by the master may not be reflected on the slave. */ * snapshot taken by the master may not be reflected on the slave. */
if (server.masterhost == NULL && !loading_aof && expiretime != -1 && expiretime < now) { if (server.masterhost == NULL && !(rdbflags&RDBFLAGS_AOF_PREAMBLE) && expiretime != -1 && expiretime < now) {
decrRefCount(key); decrRefCount(key);
decrRefCount(val); decrRefCount(val);
} else { } else {
...@@ -2243,17 +2289,17 @@ eoferr: ...@@ -2243,17 +2289,17 @@ eoferr:
* *
* If you pass an 'rsi' structure initialied with RDB_SAVE_OPTION_INIT, the * If you pass an 'rsi' structure initialied with RDB_SAVE_OPTION_INIT, the
* loading code will fiil the information fields in the structure. */ * loading code will fiil the information fields in the structure. */
int rdbLoad(char *filename, rdbSaveInfo *rsi) { int rdbLoad(char *filename, rdbSaveInfo *rsi, int rdbflags) {
FILE *fp; FILE *fp;
rio rdb; rio rdb;
int retval; int retval;
if ((fp = fopen(filename,"r")) == NULL) return C_ERR; if ((fp = fopen(filename,"r")) == NULL) return C_ERR;
startLoadingFile(fp, filename); startLoadingFile(fp, filename,rdbflags);
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
retval = rdbLoadRio(&rdb,rsi,0); retval = rdbLoadRio(&rdb,rdbflags,rsi);
fclose(fp); fclose(fp);
stopLoading(); stopLoading(retval==C_OK);
return retval; return retval;
} }
......
...@@ -121,8 +121,10 @@ ...@@ -121,8 +121,10 @@
#define RDB_LOAD_PLAIN (1<<1) #define RDB_LOAD_PLAIN (1<<1)
#define RDB_LOAD_SDS (1<<2) #define RDB_LOAD_SDS (1<<2)
#define RDB_SAVE_NONE 0 /* flags on the purpose of rdb save or load */
#define RDB_SAVE_AOF_PREAMBLE (1<<0) #define RDBFLAGS_NONE 0
#define RDBFLAGS_AOF_PREAMBLE (1<<0)
#define RDBFLAGS_REPLICATION (1<<1)
int rdbSaveType(rio *rdb, unsigned char type); int rdbSaveType(rio *rdb, unsigned char type);
int rdbLoadType(rio *rdb); int rdbLoadType(rio *rdb);
...@@ -135,7 +137,7 @@ uint64_t rdbLoadLen(rio *rdb, int *isencoded); ...@@ -135,7 +137,7 @@ uint64_t rdbLoadLen(rio *rdb, int *isencoded);
int rdbLoadLenByRef(rio *rdb, int *isencoded, uint64_t *lenptr); int rdbLoadLenByRef(rio *rdb, int *isencoded, uint64_t *lenptr);
int rdbSaveObjectType(rio *rdb, robj *o); int rdbSaveObjectType(rio *rdb, robj *o);
int rdbLoadObjectType(rio *rdb); int rdbLoadObjectType(rio *rdb);
int rdbLoad(char *filename, rdbSaveInfo *rsi); int rdbLoad(char *filename, rdbSaveInfo *rsi, int rdbflags);
int rdbSaveBackground(char *filename, rdbSaveInfo *rsi); int rdbSaveBackground(char *filename, rdbSaveInfo *rsi);
int rdbSaveToSlavesSockets(rdbSaveInfo *rsi); int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
void rdbRemoveTempFile(pid_t childpid); void rdbRemoveTempFile(pid_t childpid);
...@@ -154,7 +156,8 @@ int rdbSaveBinaryDoubleValue(rio *rdb, double val); ...@@ -154,7 +156,8 @@ int rdbSaveBinaryDoubleValue(rio *rdb, double val);
int rdbLoadBinaryDoubleValue(rio *rdb, double *val); int rdbLoadBinaryDoubleValue(rio *rdb, double *val);
int rdbSaveBinaryFloatValue(rio *rdb, float val); int rdbSaveBinaryFloatValue(rio *rdb, float val);
int rdbLoadBinaryFloatValue(rio *rdb, float *val); int rdbLoadBinaryFloatValue(rio *rdb, float *val);
int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof); int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi);
int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi);
rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi); rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi);
#endif #endif
...@@ -202,7 +202,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -202,7 +202,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
} }
expiretime = -1; expiretime = -1;
startLoadingFile(fp, rdbfilename); startLoadingFile(fp, rdbfilename, RDBFLAGS_NONE);
while(1) { while(1) {
robj *key, *val; robj *key, *val;
...@@ -316,7 +316,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -316,7 +316,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
} }
if (closefile) fclose(fp); if (closefile) fclose(fp);
stopLoading(); stopLoading(1);
return 0; return 0;
eoferr: /* unexpected end of file is handled here with a fatal exit */ eoferr: /* unexpected end of file is handled here with a fatal exit */
...@@ -327,7 +327,7 @@ eoferr: /* unexpected end of file is handled here with a fatal exit */ ...@@ -327,7 +327,7 @@ eoferr: /* unexpected end of file is handled here with a fatal exit */
} }
err: err:
if (closefile) fclose(fp); if (closefile) fclose(fp);
stopLoading(); stopLoading(0);
return 1; return 1;
} }
......
...@@ -185,6 +185,8 @@ typedef uint64_t RedisModuleTimerID; ...@@ -185,6 +185,8 @@ typedef uint64_t RedisModuleTimerID;
#define REDISMODULE_EVENT_REPLICA_CHANGE 6 #define REDISMODULE_EVENT_REPLICA_CHANGE 6
#define REDISMODULE_EVENT_MASTER_LINK_CHANGE 7 #define REDISMODULE_EVENT_MASTER_LINK_CHANGE 7
#define REDISMODULE_EVENT_CRON_LOOP 8 #define REDISMODULE_EVENT_CRON_LOOP 8
#define REDISMODULE_EVENT_MODULE_CHANGE 9
#define REDISMODULE_EVENT_LOADING_PROGRESS 10
typedef struct RedisModuleEvent { typedef struct RedisModuleEvent {
uint64_t id; /* REDISMODULE_EVENT_... defines. */ uint64_t id; /* REDISMODULE_EVENT_... defines. */
...@@ -230,18 +232,28 @@ static const RedisModuleEvent ...@@ -230,18 +232,28 @@ static const RedisModuleEvent
RedisModuleEvent_MasterLinkChange = { RedisModuleEvent_MasterLinkChange = {
REDISMODULE_EVENT_MASTER_LINK_CHANGE, REDISMODULE_EVENT_MASTER_LINK_CHANGE,
1 1
},
RedisModuleEvent_ModuleChange = {
REDISMODULE_EVENT_MODULE_CHANGE,
1
},
RedisModuleEvent_LoadingProgress = {
REDISMODULE_EVENT_LOADING_PROGRESS,
1
}; };
/* Those are values that are used for the 'subevent' callback argument. */ /* Those are values that are used for the 'subevent' callback argument. */
#define REDISMODULE_SUBEVENT_PERSISTENCE_RDB_START 0 #define REDISMODULE_SUBEVENT_PERSISTENCE_RDB_START 0
#define REDISMODULE_SUBEVENT_PERSISTENCE_RDB_END 1 #define REDISMODULE_SUBEVENT_PERSISTENCE_AOF_START 1
#define REDISMODULE_SUBEVENT_PERSISTENCE_AOF_START 2 #define REDISMODULE_SUBEVENT_PERSISTENCE_SYNC_RDB_START 2
#define REDISMODULE_SUBEVENT_PERSISTENCE_AOF_END 3 #define REDISMODULE_SUBEVENT_PERSISTENCE_ENDED 3
#define REDISMODULE_SUBEVENT_PERSISTENCE_FAILED 4
#define REDISMODULE_SUBEVENT_LOADING_RDB_START 0 #define REDISMODULE_SUBEVENT_LOADING_RDB_START 0
#define REDISMODULE_SUBEVENT_LOADING_RDB_END 1 #define REDISMODULE_SUBEVENT_LOADING_AOF_START 1
#define REDISMODULE_SUBEVENT_LOADING_AOF_START 2 #define REDISMODULE_SUBEVENT_LOADING_REPL_START 2
#define REDISMODULE_SUBEVENT_LOADING_AOF_END 3 #define REDISMODULE_SUBEVENT_LOADING_ENDED 3
#define REDISMODULE_SUBEVENT_LOADING_FAILED 4
#define REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED 0 #define REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED 0
#define REDISMODULE_SUBEVENT_CLIENT_CHANGE_DISCONNECTED 1 #define REDISMODULE_SUBEVENT_CLIENT_CHANGE_DISCONNECTED 1
...@@ -249,12 +261,21 @@ static const RedisModuleEvent ...@@ -249,12 +261,21 @@ static const RedisModuleEvent
#define REDISMODULE_SUBEVENT_MASTER_LINK_UP 0 #define REDISMODULE_SUBEVENT_MASTER_LINK_UP 0
#define REDISMODULE_SUBEVENT_MASTER_LINK_DOWN 1 #define REDISMODULE_SUBEVENT_MASTER_LINK_DOWN 1
#define REDISMODULE_SUBEVENT_REPLICA_CHANGE_CONNECTED 0 #define REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE 0
#define REDISMODULE_SUBEVENT_REPLICA_CHANGE_DISCONNECTED 1 #define REDISMODULE_SUBEVENT_REPLICA_CHANGE_OFFLINE 1
#define REDISMODULE_EVENT_REPLROLECHANGED_NOW_MASTER 0
#define REDISMODULE_EVENT_REPLROLECHANGED_NOW_REPLICA 1
#define REDISMODULE_SUBEVENT_FLUSHDB_START 0 #define REDISMODULE_SUBEVENT_FLUSHDB_START 0
#define REDISMODULE_SUBEVENT_FLUSHDB_END 1 #define REDISMODULE_SUBEVENT_FLUSHDB_END 1
#define REDISMODULE_SUBEVENT_MODULE_LOADED 0
#define REDISMODULE_SUBEVENT_MODULE_UNLOADED 1
#define REDISMODULE_SUBEVENT_LOADING_PROGRESS_RDB 0
#define REDISMODULE_SUBEVENT_LOADING_PROGRESS_AOF 1
/* RedisModuleClientInfo flags. */ /* RedisModuleClientInfo flags. */
#define REDISMODULE_CLIENTINFO_FLAG_SSL (1<<0) #define REDISMODULE_CLIENTINFO_FLAG_SSL (1<<0)
#define REDISMODULE_CLIENTINFO_FLAG_PUBSUB (1<<1) #define REDISMODULE_CLIENTINFO_FLAG_PUBSUB (1<<1)
...@@ -290,6 +311,22 @@ typedef struct RedisModuleClientInfo { ...@@ -290,6 +311,22 @@ typedef struct RedisModuleClientInfo {
#define RedisModuleClientInfo RedisModuleClientInfoV1 #define RedisModuleClientInfo RedisModuleClientInfoV1
#define REDISMODULE_REPLICATIONINFO_VERSION 1
typedef struct RedisModuleReplicationInfo {
uint64_t version; /* Not used since this structure is never passed
from the module to the core right now. Here
for future compatibility. */
int master; /* true if master, false if replica */
char *masterhost; /* master instance hostname for NOW_REPLICA */
int masterport; /* master instance port for NOW_REPLICA */
char *replid1; /* Main replication ID */
char *replid2; /* Secondary replication ID */
uint64_t repl1_offset; /* Main replication offset */
uint64_t repl2_offset; /* Offset of replid2 validity */
} RedisModuleReplicationInfoV1;
#define RedisModuleReplicationInfo RedisModuleReplicationInfoV1
#define REDISMODULE_FLUSHINFO_VERSION 1 #define REDISMODULE_FLUSHINFO_VERSION 1
typedef struct RedisModuleFlushInfo { typedef struct RedisModuleFlushInfo {
uint64_t version; /* Not used since this structure is never passed uint64_t version; /* Not used since this structure is never passed
...@@ -301,6 +338,39 @@ typedef struct RedisModuleFlushInfo { ...@@ -301,6 +338,39 @@ typedef struct RedisModuleFlushInfo {
#define RedisModuleFlushInfo RedisModuleFlushInfoV1 #define RedisModuleFlushInfo RedisModuleFlushInfoV1
#define REDISMODULE_MODULE_CHANGE_VERSION 1
typedef struct RedisModuleModuleChange {
uint64_t version; /* Not used since this structure is never passed
from the module to the core right now. Here
for future compatibility. */
const char* module_name;/* Name of module loaded or unloaded. */
int32_t module_version; /* Module version. */
} RedisModuleModuleChangeV1;
#define RedisModuleModuleChange RedisModuleModuleChangeV1
#define REDISMODULE_CRON_LOOP_VERSION 1
typedef struct RedisModuleCronLoopInfo {
uint64_t version; /* Not used since this structure is never passed
from the module to the core right now. Here
for future compatibility. */
int32_t hz; /* Approximate number of events per second. */
} RedisModuleCronLoopV1;
#define RedisModuleCronLoop RedisModuleCronLoopV1
#define REDISMODULE_LOADING_PROGRESS_VERSION 1
typedef struct RedisModuleLoadingProgressInfo {
uint64_t version; /* Not used since this structure is never passed
from the module to the core right now. Here
for future compatibility. */
int32_t hz; /* Approximate number of events per second. */
int32_t progress; /* Approximate progress between 0 and 1024, or -1
* if unknown. */
} RedisModuleLoadingProgressV1;
#define RedisModuleLoadingProgress RedisModuleLoadingProgressV1
/* ------------------------- End of common defines ------------------------ */ /* ------------------------- End of common defines ------------------------ */
#ifndef REDISMODULE_CORE #ifndef REDISMODULE_CORE
...@@ -420,6 +490,9 @@ char *REDISMODULE_API_FUNC(RedisModule_StringDMA)(RedisModuleKey *key, size_t *l ...@@ -420,6 +490,9 @@ char *REDISMODULE_API_FUNC(RedisModule_StringDMA)(RedisModuleKey *key, size_t *l
int REDISMODULE_API_FUNC(RedisModule_StringTruncate)(RedisModuleKey *key, size_t newlen); int REDISMODULE_API_FUNC(RedisModule_StringTruncate)(RedisModuleKey *key, size_t newlen);
mstime_t REDISMODULE_API_FUNC(RedisModule_GetExpire)(RedisModuleKey *key); mstime_t REDISMODULE_API_FUNC(RedisModule_GetExpire)(RedisModuleKey *key);
int REDISMODULE_API_FUNC(RedisModule_SetExpire)(RedisModuleKey *key, mstime_t expire); int REDISMODULE_API_FUNC(RedisModule_SetExpire)(RedisModuleKey *key, mstime_t expire);
void REDISMODULE_API_FUNC(RedisModule_ResetDataset)(int restart_aof, int async);
unsigned long long REDISMODULE_API_FUNC(RedisModule_DbSize)(RedisModuleCtx *ctx);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_RandomKey)(RedisModuleCtx *ctx);
int REDISMODULE_API_FUNC(RedisModule_ZsetAdd)(RedisModuleKey *key, double score, RedisModuleString *ele, int *flagsptr); int REDISMODULE_API_FUNC(RedisModule_ZsetAdd)(RedisModuleKey *key, double score, RedisModuleString *ele, int *flagsptr);
int REDISMODULE_API_FUNC(RedisModule_ZsetIncrby)(RedisModuleKey *key, double score, RedisModuleString *ele, int *flagsptr, double *newscore); int REDISMODULE_API_FUNC(RedisModule_ZsetIncrby)(RedisModuleKey *key, double score, RedisModuleString *ele, int *flagsptr, double *newscore);
int REDISMODULE_API_FUNC(RedisModule_ZsetScore)(RedisModuleKey *key, RedisModuleString *ele, double *score); int REDISMODULE_API_FUNC(RedisModule_ZsetScore)(RedisModuleKey *key, RedisModuleString *ele, double *score);
...@@ -439,6 +512,7 @@ int REDISMODULE_API_FUNC(RedisModule_IsKeysPositionRequest)(RedisModuleCtx *ctx) ...@@ -439,6 +512,7 @@ int REDISMODULE_API_FUNC(RedisModule_IsKeysPositionRequest)(RedisModuleCtx *ctx)
void REDISMODULE_API_FUNC(RedisModule_KeyAtPos)(RedisModuleCtx *ctx, int pos); void REDISMODULE_API_FUNC(RedisModule_KeyAtPos)(RedisModuleCtx *ctx, int pos);
unsigned long long REDISMODULE_API_FUNC(RedisModule_GetClientId)(RedisModuleCtx *ctx); unsigned long long REDISMODULE_API_FUNC(RedisModule_GetClientId)(RedisModuleCtx *ctx);
int REDISMODULE_API_FUNC(RedisModule_GetClientInfoById)(void *ci, uint64_t id); int REDISMODULE_API_FUNC(RedisModule_GetClientInfoById)(void *ci, uint64_t id);
int REDISMODULE_API_FUNC(RedisModule_PublishMessage)(RedisModuleCtx *ctx, RedisModuleString *channel, RedisModuleString *message);
int REDISMODULE_API_FUNC(RedisModule_GetContextFlags)(RedisModuleCtx *ctx); 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);
...@@ -509,6 +583,9 @@ int REDISMODULE_API_FUNC(RedisModule_InfoAddFieldULongLong)(RedisModuleInfoCtx * ...@@ -509,6 +583,9 @@ int REDISMODULE_API_FUNC(RedisModule_InfoAddFieldULongLong)(RedisModuleInfoCtx *
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_SetLRUOrLFU)(RedisModuleKey *key, long long lfu_freq, long long lru_idle);
int REDISMODULE_API_FUNC(RedisModule_GetLRUOrLFU)(RedisModuleKey *key, long long *lfu_freq, long long *lru_idle); int REDISMODULE_API_FUNC(RedisModule_GetLRUOrLFU)(RedisModuleKey *key, long long *lfu_freq, long long *lru_idle);
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);
RedisModuleString *REDISMODULE_API_FUNC(RedisModule_GetBlockedClientReadyKey)(RedisModuleCtx *ctx);
/* Experimental APIs */ /* Experimental APIs */
#ifdef REDISMODULE_EXPERIMENTAL_API #ifdef REDISMODULE_EXPERIMENTAL_API
...@@ -622,6 +699,9 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -622,6 +699,9 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(StringTruncate); REDISMODULE_GET_API(StringTruncate);
REDISMODULE_GET_API(GetExpire); REDISMODULE_GET_API(GetExpire);
REDISMODULE_GET_API(SetExpire); REDISMODULE_GET_API(SetExpire);
REDISMODULE_GET_API(ResetDataset);
REDISMODULE_GET_API(DbSize);
REDISMODULE_GET_API(RandomKey);
REDISMODULE_GET_API(ZsetAdd); REDISMODULE_GET_API(ZsetAdd);
REDISMODULE_GET_API(ZsetIncrby); REDISMODULE_GET_API(ZsetIncrby);
REDISMODULE_GET_API(ZsetScore); REDISMODULE_GET_API(ZsetScore);
...@@ -708,9 +788,13 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -708,9 +788,13 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(InfoAddFieldLongLong); REDISMODULE_GET_API(InfoAddFieldLongLong);
REDISMODULE_GET_API(InfoAddFieldULongLong); REDISMODULE_GET_API(InfoAddFieldULongLong);
REDISMODULE_GET_API(GetClientInfoById); REDISMODULE_GET_API(GetClientInfoById);
REDISMODULE_GET_API(PublishMessage);
REDISMODULE_GET_API(SubscribeToServerEvent); REDISMODULE_GET_API(SubscribeToServerEvent);
REDISMODULE_GET_API(SetLRUOrLFU); REDISMODULE_GET_API(SetLRUOrLFU);
REDISMODULE_GET_API(GetLRUOrLFU); REDISMODULE_GET_API(GetLRUOrLFU);
REDISMODULE_GET_API(BlockClientOnKeys);
REDISMODULE_GET_API(SignalKeyAsReady);
REDISMODULE_GET_API(GetBlockedClientReadyKey);
#ifdef REDISMODULE_EXPERIMENTAL_API #ifdef REDISMODULE_EXPERIMENTAL_API
REDISMODULE_GET_API(GetThreadSafeContext); REDISMODULE_GET_API(GetThreadSafeContext);
......
...@@ -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;
......
...@@ -2056,6 +2056,12 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { ...@@ -2056,6 +2056,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;
} }
...@@ -3682,6 +3688,9 @@ int prepareForShutdown(int flags) { ...@@ -3682,6 +3688,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.");
...@@ -4767,7 +4776,7 @@ void loadDataFromDisk(void) { ...@@ -4767,7 +4776,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);
......
...@@ -1602,6 +1602,10 @@ ssize_t rdbSaveModulesAux(rio *rdb, int when); ...@@ -1602,6 +1602,10 @@ ssize_t rdbSaveModulesAux(rio *rdb, int when);
int moduleAllDatatypesHandleErrors(); int moduleAllDatatypesHandleErrors();
sds modulesCollectInfo(sds info, sds section, int for_crash_report, int sections); sds modulesCollectInfo(sds info, sds 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);
void moduleUnblockClient(client *c);
int moduleClientIsBlockedOnKeys(client *c);
/* Utils */ /* Utils */
long long ustime(void); long long ustime(void);
...@@ -1831,10 +1835,12 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, ...@@ -1831,10 +1835,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. */
...@@ -1843,7 +1849,6 @@ int writeCommandsDeniedByDiskError(void); ...@@ -1843,7 +1849,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 */
...@@ -1859,6 +1864,7 @@ void aofRewriteBufferReset(void); ...@@ -1859,6 +1864,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);
...@@ -2099,6 +2105,7 @@ robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o); ...@@ -2099,6 +2105,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);
......
...@@ -18,7 +18,8 @@ TEST_MODULES = \ ...@@ -18,7 +18,8 @@ TEST_MODULES = \
infotest.so \ infotest.so \
propagate.so \ propagate.so \
misc.so \ misc.so \
hooks.so hooks.so \
blockonkeys.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;
}
...@@ -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;
}
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]
}
}
...@@ -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 select 2
r flushdb r flushdb
r select 9 assert_equal [r hooks.event_last flush-start] 9
assert {[r llen flush-start] == 2} assert_equal [r hooks.event_last flush-end] 9
assert {[r llen flush-end] == 3} r flushall
assert {[r lrange flush-start 0 -1] eq {1 2}} assert_equal [r hooks.event_last flush-start] -1
assert {[r lrange flush-end 0 -1] eq {-1 1 2}} assert_equal [r hooks.event_last flush-end] -1
}
# 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