caleb-brown/redict
1
0
Fork 0
mirror of https://codeberg.org/redict/redict.git synced 2025-01-22 16:18:28 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'unstable' of github.com:/antirez/redis into unstable

Browse Source
This commit is contained in:
antirez 2019-10-28 11:09:09 +01:00
commit 8337e083b6
13 changed files with 139 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
redis.conf
View File

@ -1347,7 +1347,7 @@ notify-keyspace-events ""
# to server Gopher pages MAKE SURE TO SET A PASSWORD to the instance.
# Once a password is set:
#
# 1. The Gopher server (when enabled, not by default) will kill serve
# 1. The Gopher server (when enabled, not by default) will still serve
# content via Gopher.
# 2. However other commands cannot be called before the client will
# authenticate.
@ -1669,4 +1669,3 @@ rdb-save-incremental-fsync yes
# Maximum number of set/hash/zset/list fields that will be processed from
# the main dictionary scan
# active-defrag-max-scan-fields 1000

2
src/latency.c
View File

@ -95,7 +95,7 @@ void latencyMonitorInit(void) {
* This function is usually called via latencyAddSampleIfNeeded(), that
* is a macro that only adds the sample if the latency is higher than
* server.latency_monitor_threshold. */
void latencyAddSample(char *event, mstime_t latency) {
void latencyAddSample(const char *event, mstime_t latency) {
struct latencyTimeSeries *ts = dictFetchValue(server.latency_events,event);
time_t now = time(NULL);
int prev;

2
src/latency.h
View File

@ -62,7 +62,7 @@ struct latencyStats {
};
void latencyMonitorInit(void);
void latencyAddSample(char *event, mstime_t latency);
void latencyAddSample(const char *event, mstime_t latency);
int THPIsEnabled(void);
/* Latency monitoring macros. */

66
src/module.c
View File

@ -63,6 +63,7 @@ struct RedisModule {
int in_call; /* RM_Call() nesting level */
int in_hook; /* Hooks callback nesting level for this module (0 or 1). */
int options; /* Module options and capabilities. */
int blocked_clients; /* Count of RedisModuleBlockedClient in this module. */
RedisModuleInfoFunc info_cb; /* Callback for module to add INFO fields. */
};
typedef struct RedisModule RedisModule;
@ -1264,6 +1265,27 @@ int RM_ReplyWithArray(RedisModuleCtx *ctx, long len) {
return REDISMODULE_OK;
}
/* Reply to the client with a null array, simply null in RESP3
* null array in RESP2.
*
* The function always returns REDISMODULE_OK. */
int RM_ReplyWithNullArray(RedisModuleCtx *ctx) {
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
addReplyNullArray(c);
return REDISMODULE_OK;
}
/* Reply to the client with an empty array.
*
* The function always returns REDISMODULE_OK. */
int RM_ReplyWithEmptyArray(RedisModuleCtx *ctx) {
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
addReply(c,shared.emptyarray);
return REDISMODULE_OK;
}
/* When RedisModule_ReplyWithArray() is used with the argument
* REDISMODULE_POSTPONED_ARRAY_LEN, because we don't know beforehand the number
* of items we are going to output as elements of the array, this function
@ -1342,6 +1364,27 @@ int RM_ReplyWithString(RedisModuleCtx *ctx, RedisModuleString *str) {
return REDISMODULE_OK;
}
/* Reply with an empty string.
*
* The function always returns REDISMODULE_OK. */
int RM_ReplyWithEmptyString(RedisModuleCtx *ctx) {
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
addReplyBulkCBuffer(c, "", 0);
return REDISMODULE_OK;
}
/* Reply with a binary safe string, which should not be escaped or filtered
* taking in input a C buffer pointer and length.
*
* The function always returns REDISMODULE_OK. */
int RM_ReplyWithVerbatimString(RedisModuleCtx *ctx, const char *buf, size_t len) {
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
addReplyVerbatim(c, buf, len, "txt");
return REDISMODULE_OK;
}
/* Reply to the client with a NULL. In the RESP protocol a NULL is encoded
* as the string "$-1\r\n".
*
@ -3890,6 +3933,14 @@ void RM__Assert(const char *estr, const char *file, int line) {
_serverAssert(estr, file, line);
}
/* Allows adding event to the latency monitor to be observed by the LATENCY
* command. The call is skipped if the latency is smaller than the configured
* latency-monitor-threshold. */
void RM_LatencyAddSample(const char *event, mstime_t latency) {
if (latency >= server.latency_monitor_threshold)
latencyAddSample(event, latency);
}
/* --------------------------------------------------------------------------
* Blocking clients from modules
* -------------------------------------------------------------------------- */
@ -3961,6 +4012,7 @@ RedisModuleBlockedClient *RM_BlockClient(RedisModuleCtx *ctx, RedisModuleCmdFunc
c->bpop.module_blocked_handle = zmalloc(sizeof(RedisModuleBlockedClient));
RedisModuleBlockedClient *bc = c->bpop.module_blocked_handle;
ctx->module->blocked_clients++;
/* We need to handle the invalid operation of calling modules blocking
* commands from Lua or MULTI. We actually create an already aborted
@ -4119,6 +4171,7 @@ void moduleHandleBlockedClients(void) {
/* Free 'bc' only after unblocking the client, since it is
* referenced in the client blocking context, and must be valid
* when calling unblockClient(). */
bc->module->blocked_clients--;
zfree(bc);
/* Lock again before to iterate the loop. */
@ -6089,6 +6142,7 @@ int moduleLoad(const char *path, void **module_argv, int module_argc) {
/* Redis module loaded! Register it. */
dictAdd(modules,ctx.module->name,ctx.module);
ctx.module->blocked_clients = 0;
ctx.module->handle = handle;
serverLog(LL_NOTICE,"Module '%s' loaded from %s",ctx.module->name,path);
moduleFreeContext(&ctx);
@ -6114,6 +6168,9 @@ int moduleUnload(sds name) {
} else if (listLength(module->usedby)) {
errno = EPERM;
return REDISMODULE_ERR;
} else if (module->blocked_clients) {
errno = EAGAIN;
return REDISMODULE_ERR;
}
/* Give module a chance to clean up. */
@ -6279,6 +6336,10 @@ NULL
errmsg = "the module exports APIs used by other modules. "
"Please unload them first and try again";
break;
case EAGAIN:
errmsg = "the module has blocked clients. "
"Please wait them unblocked and try again";
break;
default:
errmsg = "operation not possible.";
break;
@ -6316,8 +6377,12 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(ReplyWithError);
REGISTER_API(ReplyWithSimpleString);
REGISTER_API(ReplyWithArray);
REGISTER_API(ReplyWithNullArray);
REGISTER_API(ReplyWithEmptyArray);
REGISTER_API(ReplySetArrayLength);
REGISTER_API(ReplyWithString);
REGISTER_API(ReplyWithEmptyString);
REGISTER_API(ReplyWithVerbatimString);
REGISTER_API(ReplyWithStringBuffer);
REGISTER_API(ReplyWithCString);
REGISTER_API(ReplyWithNull);
@ -6400,6 +6465,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(Log);
REGISTER_API(LogIOError);
REGISTER_API(_Assert);
REGISTER_API(LatencyAddSample);
REGISTER_API(StringAppendBuffer);
REGISTER_API(RetainString);
REGISTER_API(StringCompare);

8
src/rdb.c
View File

@ -2172,7 +2172,11 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
/* Read key */
if ((key = rdbLoadStringObject(rdb)) == NULL) goto eoferr;
/* Read value */
if ((val = rdbLoadObject(type,rdb,key)) == NULL) goto eoferr;
if ((val = rdbLoadObject(type,rdb,key)) == NULL) {
decrRefCount(key);
goto eoferr;
}
/* Check if the key already expired. This function is used when loading
* an RDB file from disk, either at startup, or when an RDB was
* received from the master. In the latter case, the master is
@ -2289,7 +2293,7 @@ void backgroundSaveDoneHandlerDisk(int exitcode, int bysignal) {
}
/* A background saving child (BGSAVE) terminated its work. Handle this.
* This function covers the case of RDB -> Salves socket transfers for
* This function covers the case of RDB -> Slaves socket transfers for
* diskless replication. */
void backgroundSaveDoneHandlerSocket(int exitcode, int bysignal) {
if (!bysignal && exitcode == 0) {

2
src/redis-cli.c
View File

@ -370,7 +370,7 @@ static sds percentDecode(const char *pe, size_t len) {
* URI scheme is based on the the provisional specification[1] excluding support
* for query parameters. Valid URIs are:
* scheme: "redis://"
* authority: [<username> ":"] <password> "@"] [<hostname> [":" <port>]]
* authority: [[<username> ":"] <password> "@"] [<hostname> [":" <port>]]
* path: ["/" [<db>]]
*
* [1]: https://www.iana.org/assignments/uri-schemes/prov/redis */

11
src/redismodule.h
View File

@ -382,10 +382,14 @@ const char *REDISMODULE_API_FUNC(RedisModule_StringPtrLen)(const RedisModuleStri
int REDISMODULE_API_FUNC(RedisModule_ReplyWithError)(RedisModuleCtx *ctx, const char *err);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithSimpleString)(RedisModuleCtx *ctx, const char *msg);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithArray)(RedisModuleCtx *ctx, long len);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithNullArray)(RedisModuleCtx *ctx);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithEmptyArray)(RedisModuleCtx *ctx);
void REDISMODULE_API_FUNC(RedisModule_ReplySetArrayLength)(RedisModuleCtx *ctx, long len);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithStringBuffer)(RedisModuleCtx *ctx, const char *buf, size_t len);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithCString)(RedisModuleCtx *ctx, const char *buf);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithString)(RedisModuleCtx *ctx, RedisModuleString *str);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithEmptyString)(RedisModuleCtx *ctx);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithVerbatimString)(RedisModuleCtx *ctx, const char *buf, size_t len);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithNull)(RedisModuleCtx *ctx);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithDouble)(RedisModuleCtx *ctx, double d);
int REDISMODULE_API_FUNC(RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, RedisModuleCallReply *reply);
@ -446,6 +450,7 @@ float REDISMODULE_API_FUNC(RedisModule_LoadFloat)(RedisModuleIO *io);
void REDISMODULE_API_FUNC(RedisModule_Log)(RedisModuleCtx *ctx, const char *level, const char *fmt, ...);
void REDISMODULE_API_FUNC(RedisModule_LogIOError)(RedisModuleIO *io, const char *levelstr, const char *fmt, ...);
void REDISMODULE_API_FUNC(RedisModule__Assert)(const char *estr, const char *file, int line);
void REDISMODULE_API_FUNC(RedisModule_LatencyAddSample)(const char *event, mstime_t latency);
int REDISMODULE_API_FUNC(RedisModule_StringAppendBuffer)(RedisModuleCtx *ctx, RedisModuleString *str, const char *buf, size_t len);
void REDISMODULE_API_FUNC(RedisModule_RetainString)(RedisModuleCtx *ctx, RedisModuleString *str);
int REDISMODULE_API_FUNC(RedisModule_StringCompare)(RedisModuleString *a, RedisModuleString *b);
@ -552,14 +557,17 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(ReplyWithError);
REDISMODULE_GET_API(ReplyWithSimpleString);
REDISMODULE_GET_API(ReplyWithArray);
REDISMODULE_GET_API(ReplyWithNullArray);
REDISMODULE_GET_API(ReplyWithEmptyArray);
REDISMODULE_GET_API(ReplySetArrayLength);
REDISMODULE_GET_API(ReplyWithStringBuffer);
REDISMODULE_GET_API(ReplyWithCString);
REDISMODULE_GET_API(ReplyWithString);
REDISMODULE_GET_API(ReplyWithEmptyString);
REDISMODULE_GET_API(ReplyWithVerbatimString);
REDISMODULE_GET_API(ReplyWithNull);
REDISMODULE_GET_API(ReplyWithCallReply);
REDISMODULE_GET_API(ReplyWithDouble);
REDISMODULE_GET_API(ReplySetArrayLength);
REDISMODULE_GET_API(GetSelectedDb);
REDISMODULE_GET_API(SelectDb);
REDISMODULE_GET_API(OpenKey);
@ -637,6 +645,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(Log);
REDISMODULE_GET_API(LogIOError);
REDISMODULE_GET_API(_Assert);
REDISMODULE_GET_API(LatencyAddSample);
REDISMODULE_GET_API(StringAppendBuffer);
REDISMODULE_GET_API(RetainString);
REDISMODULE_GET_API(StringCompare);

6
src/replication.c
View File

@ -256,7 +256,7 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
while((ln = listNext(&li))) {
client *slave = ln->value;
/* Don't feed slaves that are still waiting for BGSAVE to start */
/* Don't feed slaves that are still waiting for BGSAVE to start. */
if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue;
/* Feed slaves that are waiting for the initial SYNC (so these commands
@ -295,7 +295,7 @@ void replicationFeedSlavesFromMasterStream(list *slaves, char *buf, size_t bufle
while((ln = listNext(&li))) {
client *slave = ln->value;
/* Don't feed slaves that are still waiting for BGSAVE to start */
/* Don't feed slaves that are still waiting for BGSAVE to start. */
if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue;
addReplyProto(slave,buf,buflen);
}
@ -585,7 +585,7 @@ int startBgsaveForReplication(int mincapa) {
}
/* If we failed to BGSAVE, remove the slaves waiting for a full
* resynchorinization from the list of slaves, inform them with
* resynchronization from the list of slaves, inform them with
* an error about what happened, close the connection ASAP. */
if (retval == C_ERR) {
serverLog(LL_WARNING,"BGSAVE for replication failed");

4
src/server.c
View File

@ -2104,7 +2104,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
/* Check if there are clients unblocked by modules that implement
* blocking commands. */
moduleHandleBlockedClients();
if (moduleCount()) moduleHandleBlockedClients();
/* Try to process pending commands for clients that were just unblocked. */
if (listLength(server.unblocked_clients))
@ -3208,7 +3208,7 @@ void preventCommandReplication(client *c) {
* CMD_CALL_STATS Populate command stats.
* CMD_CALL_PROPAGATE_AOF Append command to AOF if it modified the dataset
* or if the client flags are forcing propagation.
* CMD_CALL_PROPAGATE_REPL Send command to salves if it modified the dataset
* CMD_CALL_PROPAGATE_REPL Send command to slaves if it modified the dataset
* or if the client flags are forcing propagation.
* CMD_CALL_PROPAGATE Alias for PROPAGATE_AOF|PROPAGATE_REPL.
* CMD_CALL_FULL Alias for SLOWLOG|STATS|PROPAGATE.

39
tests/modules/Makefile
View File

@ -11,34 +11,25 @@ else
SHOBJ_LDFLAGS ?= -bundle -undefined dynamic_lookup
endif
.SUFFIXES: .c .so .xo .o
TEST_MODULES = \
commandfilter.so \
testrdb.so \
fork.so \
infotest.so \
propagate.so \
hooks.so
all: commandfilter.so testrdb.so fork.so infotest.so propagate.so hooks.so
.PHONY: all
.c.xo:
all: $(TEST_MODULES)
%.xo: %.c ../../src/redismodule.h
$(CC) -I../../src $(CFLAGS) $(SHOBJ_CFLAGS) -fPIC -c $< -o $@
commandfilter.xo: ../../src/redismodule.h
fork.xo: ../../src/redismodule.h
testrdb.xo: ../../src/redismodule.h
infotest.xo: ../../src/redismodule.h
propagate.xo: ../../src/redismodule.h
hooks.xo: ../../src/redismodule.h
commandfilter.so: commandfilter.xo
%.so: %.xo
$(LD) -o $@ $< $(SHOBJ_LDFLAGS) $(LIBS) -lc
fork.so: fork.xo
$(LD) -o $@ $< $(SHOBJ_LDFLAGS) $(LIBS) -lc
.PHONY: clean
testrdb.so: testrdb.xo
$(LD) -o $@ $< $(SHOBJ_LDFLAGS) $(LIBS) -lc
infotest.so: infotest.xo
$(LD) -o $@ $< $(SHOBJ_LDFLAGS) $(LIBS) -lc
propagate.so: propagate.xo
$(LD) -o $@ $< $(SHOBJ_LDFLAGS) $(LIBS) -lc
hooks.so: hooks.xo
$(LD) -o $@ $< $(SHOBJ_LDFLAGS) $(LIBS) -lc
clean:
rm -f $(TEST_MODULES) $(TEST_MODULES:.so=.xo)

5
tests/modules/commandfilter.c
View File

@ -147,3 +147,8 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
return REDISMODULE_OK;
}
int RedisModule_OnUnload(RedisModuleCtx *ctx) {
RedisModule_FreeString(ctx, log_key_name);
return REDISMODULE_OK;
}

8
tests/modules/testrdb.c
View File

@ -238,3 +238,11 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
return REDISMODULE_OK;
}
int RedisModule_OnUnload(RedisModuleCtx *ctx) {
if (before_str)
RedisModule_FreeString(ctx, before_str);
if (after_str)
RedisModule_FreeString(ctx, after_str);
return REDISMODULE_OK;
}

37
tests/unit/moduleapi/testrdb.tcl
View File

@ -1,57 +1,48 @@
set testmodule [file normalize tests/modules/testrdb.so]
proc restart_and_wait {} {
catch {
r debug restart
}
# wait for the server to come back up
set retry 50
while {$retry} {
if {[catch { r ping }]} {
after 100
} else {
break
}
incr retry -1
}
}
tags "modules" {
start_server [list overrides [list loadmodule "$testmodule"]] {
test {modules are able to persist types} {
start_server [list overrides [list loadmodule "$testmodule"]] {
r testrdb.set.key key1 value1
assert_equal "value1" [r testrdb.get.key key1]
r debug reload
assert_equal "value1" [r testrdb.get.key key1]
}
}
test {modules global are lost without aux} {
set server_path [tmpdir "server.module-testrdb"]
start_server [list overrides [list loadmodule "$testmodule" "dir" $server_path]] {
r testrdb.set.before global1
assert_equal "global1" [r testrdb.get.before]
restart_and_wait
}
start_server [list overrides [list loadmodule "$testmodule" "dir" $server_path]] {
assert_equal "" [r testrdb.get.before]
}
}
start_server [list overrides [list loadmodule "$testmodule 2"]] {
test {modules are able to persist globals before and after} {
set server_path [tmpdir "server.module-testrdb"]
start_server [list overrides [list loadmodule "$testmodule 2" "dir" $server_path]] {
r testrdb.set.before global1
r testrdb.set.after global2
assert_equal "global1" [r testrdb.get.before]
assert_equal "global2" [r testrdb.get.after]
restart_and_wait
}
start_server [list overrides [list loadmodule "$testmodule 2" "dir" $server_path]] {
assert_equal "global1" [r testrdb.get.before]
assert_equal "global2" [r testrdb.get.after]
}
}
start_server [list overrides [list loadmodule "$testmodule 1"]] {
test {modules are able to persist globals just after} {
set server_path [tmpdir "server.module-testrdb"]
start_server [list overrides [list loadmodule "$testmodule 1" "dir" $server_path]] {
r testrdb.set.after global2
assert_equal "global2" [r testrdb.get.after]
restart_and_wait
}
start_server [list overrides [list loadmodule "$testmodule 1" "dir" $server_path]] {
assert_equal "global2" [r testrdb.get.after]
}
}