Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
704fc1509b | ||
|
|
3beb18b4f7 | ||
|
|
8101b5b9e5 | ||
|
|
a1978ce04c | ||
|
|
5123e2b3a1 | ||
|
|
687850a612 | ||
|
|
d4555a6e38 | ||
|
|
fc8e4804b2 |
No files matched your search
@@ -37,5 +37,6 @@ $TCLSH tests/test_helper.tcl \
|
||||
--single unit/moduleapi/hash \
|
||||
--single unit/moduleapi/zset \
|
||||
--single unit/moduleapi/stream \
|
||||
--single unit/moduleapi/load \
|
||||
--config server-threads 3 \
|
||||
"${@}"
|
||||
+5
-5
@@ -716,11 +716,6 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
|
||||
}
|
||||
}
|
||||
|
||||
if (eventLoop->flags & AE_DONT_WAIT) {
|
||||
tv.tv_sec = tv.tv_usec = 0;
|
||||
tvp = &tv;
|
||||
}
|
||||
|
||||
if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP) {
|
||||
std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
|
||||
if (!(eventLoop->beforesleepFlags & AE_SLEEP_THREADSAFE)) {
|
||||
@@ -731,6 +726,11 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
|
||||
eventLoop->beforesleep(eventLoop);
|
||||
}
|
||||
|
||||
if (eventLoop->flags & AE_DONT_WAIT) {
|
||||
tv.tv_sec = tv.tv_usec = 0;
|
||||
tvp = &tv;
|
||||
}
|
||||
|
||||
/* Call the multiplexing API, will return only on timeout or when
|
||||
* some event fires. */
|
||||
numevents = aeApiPoll(eventLoop, tvp);
|
||||
|
||||
+5
-5
@@ -285,7 +285,7 @@ int startAppendOnly(void) {
|
||||
strerror(errno));
|
||||
return C_ERR;
|
||||
}
|
||||
if (hasActiveChildProcess() && g_pserver->child_type != CHILD_TYPE_AOF) {
|
||||
if (hasActiveChildProcessOrBGSave() && g_pserver->child_type != CHILD_TYPE_AOF) {
|
||||
g_pserver->aof_rewrite_scheduled = 1;
|
||||
serverLog(LL_WARNING,"AOF was enabled but there is already another background operation. An AOF background was scheduled to start when possible.");
|
||||
} else {
|
||||
@@ -438,7 +438,7 @@ void flushAppendOnlyFile(int force) {
|
||||
* useful for graphing / monitoring purposes. */
|
||||
if (sync_in_progress) {
|
||||
latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
|
||||
} else if (hasActiveChildProcess()) {
|
||||
} else if (hasActiveChildProcessOrBGSave()) {
|
||||
latencyAddSampleIfNeeded("aof-write-active-child",latency);
|
||||
} else {
|
||||
latencyAddSampleIfNeeded("aof-write-alone",latency);
|
||||
@@ -535,7 +535,7 @@ void flushAppendOnlyFile(int force) {
|
||||
try_fsync:
|
||||
/* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
|
||||
* children doing I/O in the background. */
|
||||
if (g_pserver->aof_no_fsync_on_rewrite && hasActiveChildProcess())
|
||||
if (g_pserver->aof_no_fsync_on_rewrite && hasActiveChildProcessOrBGSave())
|
||||
return;
|
||||
|
||||
/* Perform the fsync if needed. */
|
||||
@@ -1887,7 +1887,7 @@ void aofClosePipes(void) {
|
||||
int rewriteAppendOnlyFileBackground(void) {
|
||||
pid_t childpid;
|
||||
|
||||
if (hasActiveChildProcess()) return C_ERR;
|
||||
if (hasActiveChildProcessOrBGSave()) return C_ERR;
|
||||
if (aofCreatePipes() != C_OK) return C_ERR;
|
||||
if ((childpid = redisFork(CHILD_TYPE_AOF)) == 0) {
|
||||
char tmpfile[256];
|
||||
@@ -1930,7 +1930,7 @@ int rewriteAppendOnlyFileBackground(void) {
|
||||
void bgrewriteaofCommand(client *c) {
|
||||
if (g_pserver->child_type == CHILD_TYPE_AOF) {
|
||||
addReplyError(c,"Background append only file rewriting already in progress");
|
||||
} else if (hasActiveChildProcess()) {
|
||||
} else if (hasActiveChildProcessOrBGSave()) {
|
||||
g_pserver->aof_rewrite_scheduled = 1;
|
||||
addReplyStatus(c,"Background append only file rewriting scheduled");
|
||||
} else if (rewriteAppendOnlyFileBackground() == C_OK) {
|
||||
|
||||
@@ -40,6 +40,7 @@
|
||||
#endif
|
||||
|
||||
const char *KEYDB_SET_VERSION = KEYDB_REAL_VERSION;
|
||||
size_t g_semiOrderedSetTargetBucketSize = 0; // Its a header only class so nowhere else for this to go
|
||||
|
||||
/*-----------------------------------------------------------------------------
|
||||
* Config file name-value maps.
|
||||
@@ -2910,6 +2911,7 @@ standardConfig configs[] = {
|
||||
createBoolConfig("allow-write-during-load", NULL, MODIFIABLE_CONFIG, g_pserver->fWriteDuringActiveLoad, 0, NULL, NULL),
|
||||
createBoolConfig("force-backlog-disk-reserve", NULL, MODIFIABLE_CONFIG, cserver.force_backlog_disk, 0, NULL, NULL),
|
||||
createBoolConfig("soft-shutdown", NULL, MODIFIABLE_CONFIG, g_pserver->config_soft_shutdown, 0, NULL, NULL),
|
||||
createSizeTConfig("semi-ordered-set-bucket-size", NULL, MODIFIABLE_CONFIG, 0, 1024, g_semiOrderedSetTargetBucketSize, 0, INTEGER_CONFIG, NULL, NULL),
|
||||
|
||||
#ifdef USE_OPENSSL
|
||||
createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, updateTLSPort), /* TCP port. */
|
||||
|
||||
+12
-4
@@ -93,7 +93,7 @@ static void lookupKeyUpdateObj(robj *val, int flags)
|
||||
/* Update the access time for the ageing algorithm.
|
||||
* Don't do it if we have a saving child, as this will trigger
|
||||
* a copy on write madness. */
|
||||
if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH))
|
||||
if (!hasActiveChildProcessOrBGSave() && !(flags & LOOKUP_NOTOUCH))
|
||||
{
|
||||
if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
|
||||
updateLFU(val);
|
||||
@@ -1059,7 +1059,7 @@ void keysCommand(client *c) {
|
||||
sds pattern = szFromObj(c->argv[1]);
|
||||
|
||||
const redisDbPersistentDataSnapshot *snapshot = nullptr;
|
||||
if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED)))
|
||||
if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED | CLIENT_DENY_BLOCKING)))
|
||||
snapshot = c->db->createSnapshot(c->mvccCheckpoint, true /* fOptional */);
|
||||
if (snapshot != nullptr)
|
||||
{
|
||||
@@ -2607,6 +2607,14 @@ void clusterStorageLoadCallback(const char *rgchkey, size_t cch, void *)
|
||||
slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/);
|
||||
}
|
||||
|
||||
void moduleLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
|
||||
if (g_pserver->cluster_enabled) {
|
||||
clusterStorageLoadCallback(rgchKey, cchKey, data);
|
||||
}
|
||||
robj *keyobj = createEmbeddedStringObject(rgchKey, cchKey);
|
||||
moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", keyobj, *(int *)data);
|
||||
}
|
||||
|
||||
void redisDb::initialize(int id)
|
||||
{
|
||||
redisDbPersistentData::initialize();
|
||||
@@ -2625,8 +2633,8 @@ void redisDb::storageProviderInitialize()
|
||||
{
|
||||
if (g_pserver->m_pstorageFactory != nullptr)
|
||||
{
|
||||
IStorageFactory::key_load_iterator itr = (g_pserver->cluster_enabled) ? clusterStorageLoadCallback : nullptr;
|
||||
this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, nullptr));
|
||||
IStorageFactory::key_load_iterator itr = moduleLoadCallback;
|
||||
this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, &id));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+8
-5
@@ -1679,17 +1679,15 @@ int rdbSaveBackground(rdbSaveInfo *rsi) {
|
||||
pthread_t child;
|
||||
long long start;
|
||||
|
||||
if (hasActiveChildProcess()) return C_ERR;
|
||||
if (hasActiveChildProcessOrBGSave()) return C_ERR;
|
||||
|
||||
g_pserver->dirty_before_bgsave = g_pserver->dirty;
|
||||
g_pserver->lastbgsave_try = time(NULL);
|
||||
openChildInfoPipe();
|
||||
|
||||
start = ustime();
|
||||
latencyStartMonitor(g_pserver->rdb_save_latency);
|
||||
|
||||
|
||||
g_pserver->stat_fork_time = ustime()-start;
|
||||
g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
|
||||
if (launchRdbSaveThread(child, rsi) != C_OK) {
|
||||
closeChildInfoPipe();
|
||||
g_pserver->lastbgsave_status = C_ERR;
|
||||
@@ -1697,6 +1695,9 @@ int rdbSaveBackground(rdbSaveInfo *rsi) {
|
||||
strerror(errno));
|
||||
return C_ERR;
|
||||
}
|
||||
|
||||
g_pserver->stat_fork_time = ustime()-start;
|
||||
g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
|
||||
latencyAddSampleIfNeeded("fork",g_pserver->stat_fork_time/1000);
|
||||
serverLog(LL_NOTICE,"Background saving started");
|
||||
g_pserver->rdb_save_time_start = time(NULL);
|
||||
@@ -3601,6 +3602,8 @@ static void backgroundSaveDoneHandlerDisk(int exitcode, bool fCancelled) {
|
||||
g_pserver->dirty = g_pserver->dirty - g_pserver->dirty_before_bgsave;
|
||||
g_pserver->lastsave = time(NULL);
|
||||
g_pserver->lastbgsave_status = C_OK;
|
||||
latencyEndMonitor(g_pserver->rdb_save_latency);
|
||||
latencyAddSampleIfNeeded("rdb-save",g_pserver->rdb_save_latency);
|
||||
} else if (!fCancelled && exitcode != 0) {
|
||||
serverLog(LL_WARNING, "Background saving error");
|
||||
g_pserver->lastbgsave_status = C_ERR;
|
||||
@@ -3782,7 +3785,7 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
|
||||
int pipefds[2];
|
||||
rdbSaveSocketThreadArgs *args = nullptr;
|
||||
|
||||
if (hasActiveChildProcess()) return C_ERR;
|
||||
if (hasActiveChildProcessOrBGSave()) return C_ERR;
|
||||
|
||||
/* Even if the previous fork child exited, don't start a new one until we
|
||||
* drained the pipe. */
|
||||
|
||||
@@ -326,6 +326,7 @@ static const RedisModuleEvent
|
||||
#define REDISMODULE_SUBEVENT_LOADING_ENDED 3
|
||||
#define REDISMODULE_SUBEVENT_LOADING_FAILED 4
|
||||
#define _REDISMODULE_SUBEVENT_LOADING_NEXT 5
|
||||
#define REDISMODULE_SUBEVENT_LOADING_FLASH_START 6
|
||||
|
||||
#define REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED 0
|
||||
#define REDISMODULE_SUBEVENT_CLIENT_CHANGE_DISCONNECTED 1
|
||||
|
||||
+2
-2
@@ -1538,7 +1538,7 @@ void syncCommand(client *c) {
|
||||
} else {
|
||||
/* We don't have a BGSAVE in progress, let's start one. Diskless
|
||||
* or disk-based mode is determined by replica's capacity. */
|
||||
if (!hasActiveChildProcess()) {
|
||||
if (!hasActiveChildProcessOrBGSave()) {
|
||||
startBgsaveForReplication(c->slave_capa);
|
||||
} else {
|
||||
serverLog(LL_NOTICE,
|
||||
@@ -4984,7 +4984,7 @@ void replicationStartPendingFork(void) {
|
||||
* In case of diskless replication, we make sure to wait the specified
|
||||
* number of seconds (according to configuration) so that other slaves
|
||||
* have the time to arrive before we start streaming. */
|
||||
if (!hasActiveChildProcess()) {
|
||||
if (!hasActiveChildProcessOrBGSave()) {
|
||||
time_t idle, max_idle = 0;
|
||||
int slaves_waiting = 0;
|
||||
int mincapa = -1;
|
||||
|
||||
@@ -27,6 +27,7 @@ namespace keydbutils
|
||||
template<>
|
||||
size_t hash(const sdsview &);
|
||||
}
|
||||
extern size_t g_semiOrderedSetTargetBucketSize;
|
||||
|
||||
template<typename T, typename T_KEY = T, bool MEMMOVE_SAFE = false>
|
||||
class semiorderedset
|
||||
@@ -41,11 +42,14 @@ class semiorderedset
|
||||
size_t idxRehash = (1ULL << bits_min);
|
||||
int cfPauseRehash = 0;
|
||||
|
||||
constexpr size_t targetElementsPerBucket()
|
||||
inline size_t targetElementsPerBucket()
|
||||
{
|
||||
// Aim for roughly 4 cache lines per bucket (determined by imperical testing)
|
||||
// lower values are faster but use more memory
|
||||
return std::max((64/sizeof(T))*8, (size_t)2);
|
||||
if (g_semiOrderedSetTargetBucketSize == 0)
|
||||
return std::max((64/sizeof(T))*8, (size_t)2);
|
||||
else
|
||||
return g_semiOrderedSetTargetBucketSize;
|
||||
}
|
||||
|
||||
public:
|
||||
|
||||
+20
-13
@@ -1706,7 +1706,7 @@ int redisDbPersistentData::incrementallyRehash() {
|
||||
* for dict.c to resize the hash tables accordingly to the fact we have an
|
||||
* active fork child running. */
|
||||
void updateDictResizePolicy(void) {
|
||||
if (!hasActiveChildProcess() || (g_pserver->FRdbSaveInProgress() && !cserver.fForkBgSave))
|
||||
if (!hasActiveChildProcess())
|
||||
dictEnableResize();
|
||||
else
|
||||
dictDisableResize();
|
||||
@@ -1725,7 +1725,11 @@ const char *strChildType(int type) {
|
||||
/* Return true if there are active children processes doing RDB saving,
|
||||
* AOF rewriting, or some side process spawned by a loaded module. */
|
||||
int hasActiveChildProcess() {
|
||||
return g_pserver->FRdbSaveInProgress() || g_pserver->child_pid != -1;
|
||||
return g_pserver->child_pid != -1;
|
||||
}
|
||||
|
||||
int hasActiveChildProcessOrBGSave() {
|
||||
return g_pserver->FRdbSaveInProgress() || hasActiveChildProcess();
|
||||
}
|
||||
|
||||
void resetChildState() {
|
||||
@@ -2074,7 +2078,7 @@ void databasesCron(bool fMainThread) {
|
||||
/* Perform hash tables rehashing if needed, but only if there are no
|
||||
* other processes saving the DB on disk. Otherwise rehashing is bad
|
||||
* as will cause a lot of copy-on-write of memory pages. */
|
||||
if (!hasActiveChildProcess() || g_pserver->FRdbSaveInProgress()) {
|
||||
if (!hasActiveChildProcess()) {
|
||||
/* We use global counters so if we stop the computation at a given
|
||||
* DB we'll be able to start from the successive in the next
|
||||
* cron loop iteration. */
|
||||
@@ -2468,14 +2472,14 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
|
||||
|
||||
/* Start a scheduled AOF rewrite if this was requested by the user while
|
||||
* a BGSAVE was in progress. */
|
||||
if (!hasActiveChildProcess() &&
|
||||
if (!hasActiveChildProcessOrBGSave() &&
|
||||
g_pserver->aof_rewrite_scheduled)
|
||||
{
|
||||
rewriteAppendOnlyFileBackground();
|
||||
}
|
||||
|
||||
/* Check if a background saving or AOF rewrite in progress terminated. */
|
||||
if (hasActiveChildProcess() || ldbPendingChildren())
|
||||
if (hasActiveChildProcessOrBGSave() || ldbPendingChildren())
|
||||
{
|
||||
run_with_period(1000) receiveChildInfo();
|
||||
checkChildrenDone();
|
||||
@@ -2517,7 +2521,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
|
||||
|
||||
/* Trigger an AOF rewrite if needed. */
|
||||
if (g_pserver->aof_state == AOF_ON &&
|
||||
!hasActiveChildProcess() &&
|
||||
!hasActiveChildProcessOrBGSave() &&
|
||||
g_pserver->aof_rewrite_perc &&
|
||||
g_pserver->aof_current_size > g_pserver->aof_rewrite_min_size)
|
||||
{
|
||||
@@ -2601,7 +2605,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
|
||||
* Note: this code must be after the replicationCron() call above so
|
||||
* make sure when refactoring this file to keep this order. This is useful
|
||||
* because we want to give priority to RDB savings for replication. */
|
||||
if (!hasActiveChildProcess() &&
|
||||
if (!hasActiveChildProcessOrBGSave() &&
|
||||
g_pserver->rdb_bgsave_scheduled &&
|
||||
(g_pserver->unixtime-g_pserver->lastbgsave_try > CONFIG_BGSAVE_RETRY_DELAY ||
|
||||
g_pserver->lastbgsave_status == C_OK))
|
||||
@@ -4064,12 +4068,6 @@ void initServer(void) {
|
||||
}
|
||||
}
|
||||
|
||||
/* We have to initialize storage providers after the cluster has been initialized */
|
||||
for (int idb = 0; idb < cserver.dbnum; ++idb)
|
||||
{
|
||||
g_pserver->db[idb]->storageProviderInitialize();
|
||||
}
|
||||
|
||||
saveMasterStatusToStorage(false); // eliminate the repl-offset field
|
||||
|
||||
/* Initialize ACL default password if it exists */
|
||||
@@ -4082,6 +4080,15 @@ void initServer(void) {
|
||||
* Thread Local Storage initialization collides with dlopen call.
|
||||
* see: https://sourceware.org/bugzilla/show_bug.cgi?id=19329 */
|
||||
void InitServerLast() {
|
||||
|
||||
/* We have to initialize storage providers after the cluster has been initialized */
|
||||
moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_FLASH_START, NULL);
|
||||
for (int idb = 0; idb < cserver.dbnum; ++idb)
|
||||
{
|
||||
g_pserver->db[idb]->storageProviderInitialize();
|
||||
}
|
||||
moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_ENDED, NULL);
|
||||
|
||||
bioInit();
|
||||
set_jemalloc_bg_thread(cserver.jemalloc_bg_thread);
|
||||
g_pserver->initial_memory_usage = zmalloc_used_memory();
|
||||
|
||||
+3
-1
@@ -2470,7 +2470,8 @@ struct redisServer {
|
||||
time_t lastbgsave_try; /* Unix time of last attempted bgsave */
|
||||
time_t rdb_save_time_last; /* Time used by last RDB save run. */
|
||||
time_t rdb_save_time_start; /* Current RDB save start time. */
|
||||
pid_t rdb_child_pid = -1; /* Used only during fork bgsave */
|
||||
mstime_t rdb_save_latency; /* Used to track end to end latency of rdb save*/
|
||||
pid_t rdb_child_pid = -1; /* Used only during fork bgsave */
|
||||
int rdb_bgsave_scheduled; /* BGSAVE when possible if true. */
|
||||
int rdb_child_type; /* Type of save by active child. */
|
||||
int lastbgsave_status; /* C_OK or C_ERR */
|
||||
@@ -3254,6 +3255,7 @@ void receiveChildInfo(void);
|
||||
void executeWithoutGlobalLock(std::function<void()> func);
|
||||
int redisFork(int type);
|
||||
int hasActiveChildProcess();
|
||||
int hasActiveChildProcessOrBGSave();
|
||||
void resetChildState();
|
||||
int isMutuallyExclusiveChildType(int type);
|
||||
|
||||
|
||||
@@ -39,7 +39,8 @@ TEST_MODULES = \
|
||||
defragtest.so \
|
||||
hash.so \
|
||||
zset.so \
|
||||
stream.so
|
||||
stream.so \
|
||||
load.so
|
||||
|
||||
|
||||
.PHONY: all
|
||||
|
||||
@@ -212,6 +212,7 @@ void loadingCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void
|
||||
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_FLASH_START: keyname = "loading-flash-start"; break;
|
||||
case REDISMODULE_SUBEVENT_LOADING_ENDED: keyname = "loading-end"; break;
|
||||
case REDISMODULE_SUBEVENT_LOADING_FAILED: keyname = "loading-failed"; break;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
/* Server hooks API example
|
||||
*
|
||||
* -----------------------------------------------------------------------------
|
||||
*
|
||||
* Copyright (c) 2019, Salvatore Sanfilippo <antirez at gmail dot com>
|
||||
* 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.
|
||||
*/
|
||||
|
||||
#define REDISMODULE_EXPERIMENTAL_API
|
||||
#include "redismodule.h"
|
||||
|
||||
size_t count, finalCount;
|
||||
|
||||
/* Client state change callback. */
|
||||
void loadCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data) {
|
||||
REDISMODULE_NOT_USED(ctx);
|
||||
REDISMODULE_NOT_USED(e);
|
||||
REDISMODULE_NOT_USED(data);
|
||||
|
||||
if (sub == REDISMODULE_SUBEVENT_LOADING_FLASH_START || sub == REDISMODULE_SUBEVENT_LOADING_RDB_START || sub == REDISMODULE_SUBEVENT_LOADING_AOF_START || sub == REDISMODULE_SUBEVENT_LOADING_REPL_START) {
|
||||
count = 0;
|
||||
finalCount = 0;
|
||||
} else if (sub == REDISMODULE_SUBEVENT_LOADING_ENDED) {
|
||||
finalCount = count;
|
||||
}
|
||||
}
|
||||
|
||||
int loadKeyCallback(RedisModuleCtx *ctx, int type, const char *event, RedisModuleString *key) {
|
||||
REDISMODULE_NOT_USED(ctx);
|
||||
REDISMODULE_NOT_USED(type);
|
||||
REDISMODULE_NOT_USED(event);
|
||||
|
||||
const char *keyname = RedisModule_StringPtrLen(key, NULL);
|
||||
|
||||
RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_NOTICE, "Loaded key: %s", keyname);
|
||||
|
||||
count++;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int LoadCount_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
||||
REDISMODULE_NOT_USED(argv);
|
||||
RedisModule_AutoMemory(ctx); /* Use automatic memory management. */
|
||||
|
||||
if (argc != 1) return RedisModule_WrongArity(ctx);
|
||||
|
||||
RedisModule_ReplyWithLongLong(ctx, finalCount);
|
||||
|
||||
return REDISMODULE_OK;
|
||||
}
|
||||
|
||||
/* This function must be present on each Redis module. It is used in order to
|
||||
* register the commands into the Redis server. */
|
||||
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
||||
REDISMODULE_NOT_USED(argv);
|
||||
REDISMODULE_NOT_USED(argc);
|
||||
|
||||
if (RedisModule_Init(ctx,"load",1,REDISMODULE_APIVER_1)
|
||||
== REDISMODULE_ERR) return REDISMODULE_ERR;
|
||||
|
||||
RedisModule_SubscribeToServerEvent(ctx,
|
||||
RedisModuleEvent_Loading, loadCallback);
|
||||
RedisModule_SubscribeToKeyspaceEvents(ctx,
|
||||
REDISMODULE_NOTIFY_LOADED, loadKeyCallback);
|
||||
|
||||
if (RedisModule_CreateCommand(ctx, "load.count",
|
||||
LoadCount_RedisCommand,"readonly",1,1,1) == REDISMODULE_ERR)
|
||||
return REDISMODULE_ERR;
|
||||
return REDISMODULE_OK;
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
set testmodule [file normalize tests/modules/load.so]
|
||||
|
||||
if {$::flash_enabled} {
|
||||
start_server [list tags [list "modules"] overrides [list storage-provider {flash ./rocks.db.master} databases 256 loadmodule $testmodule]] {
|
||||
test "Module is notified of keys loaded from flash" {
|
||||
r flushall
|
||||
r set foo bar
|
||||
r set bar foo
|
||||
r set foobar barfoo
|
||||
assert_equal [r load.count] 0
|
||||
r debug reload
|
||||
assert_equal [r load.count] 3
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -43,6 +43,10 @@ start_server {tags {"scripting"}} {
|
||||
r eval {return redis.call('get',KEYS[1])} 1 mykey
|
||||
} {myval}
|
||||
|
||||
test { EVAL - keys command works? } {
|
||||
r eval {return redis.call('keys', 'test')} 0
|
||||
}
|
||||
|
||||
test {EVALSHA - Can we call a SHA1 if already defined?} {
|
||||
r evalsha fd758d1589d044dd850a6f05d52f2eefd27f033f 1 mykey
|
||||
} {myval}
|
||||
|
||||
Reference in new issue
Block a user