From 317c0644ccf108aa23ef3fd8358bd66c2840bfc0 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 14 Apr 2024 15:40:54 +0200 Subject: Adding upstream version 5:7.2.4. Signed-off-by: Daniel Baumann --- tests/modules/propagate.c | 403 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 403 insertions(+) create mode 100644 tests/modules/propagate.c (limited to 'tests/modules/propagate.c') diff --git a/tests/modules/propagate.c b/tests/modules/propagate.c new file mode 100644 index 0000000..d5132a5 --- /dev/null +++ b/tests/modules/propagate.c @@ -0,0 +1,403 @@ +/* This module is used to test the propagation (replication + AOF) of + * commands, via the RedisModule_Replicate() interface, in asynchronous + * contexts, such as callbacks not implementing commands, and thread safe + * contexts. + * + * We create a timer callback and a threads using a thread safe context. + * Using both we try to propagate counters increments, and later we check + * if the replica contains the changes as expected. + * + * ----------------------------------------------------------------------------- + * + * Copyright (c) 2019, Salvatore Sanfilippo + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * * Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of Redis nor the names of its contributors may be used + * to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + +#include "redismodule.h" +#include +#include + +#define UNUSED(V) ((void) V) + +RedisModuleCtx *detached_ctx = NULL; + +static int KeySpace_NotificationGeneric(RedisModuleCtx *ctx, int type, const char *event, RedisModuleString *key) { + REDISMODULE_NOT_USED(type); + REDISMODULE_NOT_USED(event); + REDISMODULE_NOT_USED(key); + + RedisModuleCallReply* rep = RedisModule_Call(ctx, "INCR", "c!", "notifications"); + RedisModule_FreeCallReply(rep); + + return REDISMODULE_OK; +} + +/* Timer callback. */ +void timerHandler(RedisModuleCtx *ctx, void *data) { + REDISMODULE_NOT_USED(ctx); + REDISMODULE_NOT_USED(data); + + static int times = 0; + + RedisModule_Replicate(ctx,"INCR","c","timer"); + times++; + + if (times < 3) + RedisModule_CreateTimer(ctx,100,timerHandler,NULL); + else + times = 0; +} + +int propagateTestTimerCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + + RedisModuleTimerID timer_id = + RedisModule_CreateTimer(ctx,100,timerHandler,NULL); + REDISMODULE_NOT_USED(timer_id); + + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +/* Timer callback. */ +void timerNestedHandler(RedisModuleCtx *ctx, void *data) { + int repl = (long long)data; + + /* The goal is the trigger a module command that calls RM_Replicate + * in order to test MULTI/EXEC structure */ + RedisModule_Replicate(ctx,"INCRBY","cc","timer-nested-start","1"); + RedisModuleCallReply *reply = RedisModule_Call(ctx,"propagate-test.nested", repl? "!" : ""); + RedisModule_FreeCallReply(reply); + reply = RedisModule_Call(ctx, "INCR", repl? "c!" : "c", "timer-nested-middle"); + RedisModule_FreeCallReply(reply); + RedisModule_Replicate(ctx,"INCRBY","cc","timer-nested-end","1"); +} + +int propagateTestTimerNestedCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + + RedisModuleTimerID timer_id = + RedisModule_CreateTimer(ctx,100,timerNestedHandler,(void*)0); + REDISMODULE_NOT_USED(timer_id); + + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +int propagateTestTimerNestedReplCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + + RedisModuleTimerID timer_id = + RedisModule_CreateTimer(ctx,100,timerNestedHandler,(void*)1); + REDISMODULE_NOT_USED(timer_id); + + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +void timerHandlerMaxmemory(RedisModuleCtx *ctx, void *data) { + REDISMODULE_NOT_USED(ctx); + REDISMODULE_NOT_USED(data); + + RedisModuleCallReply *reply = RedisModule_Call(ctx,"SETEX","ccc!","timer-maxmemory-volatile-start","100","1"); + RedisModule_FreeCallReply(reply); + reply = RedisModule_Call(ctx, "CONFIG", "ccc!", "SET", "maxmemory", "1"); + RedisModule_FreeCallReply(reply); + + RedisModule_Replicate(ctx, "INCR", "c", "timer-maxmemory-middle"); + + reply = RedisModule_Call(ctx,"SETEX","ccc!","timer-maxmemory-volatile-end","100","1"); + RedisModule_FreeCallReply(reply); +} + +int propagateTestTimerMaxmemoryCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + + RedisModuleTimerID timer_id = + RedisModule_CreateTimer(ctx,100,timerHandlerMaxmemory,(void*)1); + REDISMODULE_NOT_USED(timer_id); + + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +void timerHandlerEval(RedisModuleCtx *ctx, void *data) { + REDISMODULE_NOT_USED(ctx); + REDISMODULE_NOT_USED(data); + + RedisModuleCallReply *reply = RedisModule_Call(ctx,"INCRBY","cc!","timer-eval-start","1"); + RedisModule_FreeCallReply(reply); + reply = RedisModule_Call(ctx, "EVAL", "cccc!", "redis.call('set',KEYS[1],ARGV[1])", "1", "foo", "bar"); + RedisModule_FreeCallReply(reply); + + RedisModule_Replicate(ctx, "INCR", "c", "timer-eval-middle"); + + reply = RedisModule_Call(ctx,"INCRBY","cc!","timer-eval-end","1"); + RedisModule_FreeCallReply(reply); +} + +int propagateTestTimerEvalCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + + RedisModuleTimerID timer_id = + RedisModule_CreateTimer(ctx,100,timerHandlerEval,(void*)1); + REDISMODULE_NOT_USED(timer_id); + + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +/* The thread entry point. */ +void *threadMain(void *arg) { + REDISMODULE_NOT_USED(arg); + RedisModuleCtx *ctx = RedisModule_GetThreadSafeContext(NULL); + RedisModule_SelectDb(ctx,9); /* Tests ran in database number 9. */ + for (int i = 0; i < 3; i++) { + RedisModule_ThreadSafeContextLock(ctx); + RedisModule_Replicate(ctx,"INCR","c","a-from-thread"); + RedisModuleCallReply *reply = RedisModule_Call(ctx,"INCR","c!","thread-call"); + RedisModule_FreeCallReply(reply); + RedisModule_Replicate(ctx,"INCR","c","b-from-thread"); + RedisModule_ThreadSafeContextUnlock(ctx); + } + RedisModule_FreeThreadSafeContext(ctx); + return NULL; +} + +int propagateTestThreadCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + + pthread_t tid; + if (pthread_create(&tid,NULL,threadMain,NULL) != 0) + return RedisModule_ReplyWithError(ctx,"-ERR Can't start thread"); + REDISMODULE_NOT_USED(tid); + + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +/* The thread entry point. */ +void *threadDetachedMain(void *arg) { + REDISMODULE_NOT_USED(arg); + RedisModule_SelectDb(detached_ctx,9); /* Tests ran in database number 9. */ + + RedisModule_ThreadSafeContextLock(detached_ctx); + RedisModule_Replicate(detached_ctx,"INCR","c","thread-detached-before"); + RedisModuleCallReply *reply = RedisModule_Call(detached_ctx,"INCR","c!","thread-detached-1"); + RedisModule_FreeCallReply(reply); + reply = RedisModule_Call(detached_ctx,"INCR","c!","thread-detached-2"); + RedisModule_FreeCallReply(reply); + RedisModule_Replicate(detached_ctx,"INCR","c","thread-detached-after"); + RedisModule_ThreadSafeContextUnlock(detached_ctx); + + return NULL; +} + +int propagateTestDetachedThreadCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + + pthread_t tid; + if (pthread_create(&tid,NULL,threadDetachedMain,NULL) != 0) + return RedisModule_ReplyWithError(ctx,"-ERR Can't start thread"); + REDISMODULE_NOT_USED(tid); + + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +int propagateTestSimpleCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + + /* Replicate two commands to test MULTI/EXEC wrapping. */ + RedisModule_Replicate(ctx,"INCR","c","counter-1"); + RedisModule_Replicate(ctx,"INCR","c","counter-2"); + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +int propagateTestMixedCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + RedisModuleCallReply *reply; + + /* This test mixes multiple propagation systems. */ + reply = RedisModule_Call(ctx, "INCR", "c!", "using-call"); + RedisModule_FreeCallReply(reply); + + RedisModule_Replicate(ctx,"INCR","c","counter-1"); + RedisModule_Replicate(ctx,"INCR","c","counter-2"); + + reply = RedisModule_Call(ctx, "INCR", "c!", "after-call"); + RedisModule_FreeCallReply(reply); + + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +int propagateTestNestedCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + RedisModuleCallReply *reply; + + /* This test mixes multiple propagation systems. */ + reply = RedisModule_Call(ctx, "INCR", "c!", "using-call"); + RedisModule_FreeCallReply(reply); + + reply = RedisModule_Call(ctx,"propagate-test.simple", "!"); + RedisModule_FreeCallReply(reply); + + RedisModule_Replicate(ctx,"INCR","c","counter-3"); + RedisModule_Replicate(ctx,"INCR","c","counter-4"); + + reply = RedisModule_Call(ctx, "INCR", "c!", "after-call"); + RedisModule_FreeCallReply(reply); + + reply = RedisModule_Call(ctx, "INCR", "c!", "before-call-2"); + RedisModule_FreeCallReply(reply); + + reply = RedisModule_Call(ctx, "keyspace.incr_case1", "c!", "asdf"); /* Propagates INCR */ + RedisModule_FreeCallReply(reply); + + reply = RedisModule_Call(ctx, "keyspace.del_key_copy", "c!", "asdf"); /* Propagates DEL */ + RedisModule_FreeCallReply(reply); + + reply = RedisModule_Call(ctx, "INCR", "c!", "after-call-2"); + RedisModule_FreeCallReply(reply); + + RedisModule_ReplyWithSimpleString(ctx,"OK"); + return REDISMODULE_OK; +} + +int propagateTestIncr(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) +{ + REDISMODULE_NOT_USED(argc); + RedisModuleCallReply *reply; + + /* This test propagates the module command, not the INCR it executes. */ + reply = RedisModule_Call(ctx, "INCR", "s", argv[1]); + RedisModule_ReplyWithCallReply(ctx,reply); + RedisModule_FreeCallReply(reply); + RedisModule_ReplicateVerbatim(ctx); + return REDISMODULE_OK; +} + +int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) { + REDISMODULE_NOT_USED(argv); + REDISMODULE_NOT_USED(argc); + + if (RedisModule_Init(ctx,"propagate-test",1,REDISMODULE_APIVER_1) + == REDISMODULE_ERR) return REDISMODULE_ERR; + + detached_ctx = RedisModule_GetDetachedThreadSafeContext(ctx); + + if (RedisModule_SubscribeToKeyspaceEvents(ctx, REDISMODULE_NOTIFY_ALL, KeySpace_NotificationGeneric) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.timer", + propagateTestTimerCommand, + "",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.timer-nested", + propagateTestTimerNestedCommand, + "",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.timer-nested-repl", + propagateTestTimerNestedReplCommand, + "",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.timer-maxmemory", + propagateTestTimerMaxmemoryCommand, + "",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.timer-eval", + propagateTestTimerEvalCommand, + "",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.thread", + propagateTestThreadCommand, + "",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.detached-thread", + propagateTestDetachedThreadCommand, + "",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.simple", + propagateTestSimpleCommand, + "",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.mixed", + propagateTestMixedCommand, + "write",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.nested", + propagateTestNestedCommand, + "write",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + if (RedisModule_CreateCommand(ctx,"propagate-test.incr", + propagateTestIncr, + "write",1,1,1) == REDISMODULE_ERR) + return REDISMODULE_ERR; + + return REDISMODULE_OK; +} + +int RedisModule_OnUnload(RedisModuleCtx *ctx) { + UNUSED(ctx); + + if (detached_ctx) + RedisModule_FreeThreadSafeContext(detached_ctx); + + return REDISMODULE_OK; +} -- cgit v1.2.3