Compare commits
11
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f3d36f1645 | ||
|
|
afa2201688 | ||
|
|
d05b27c4ed | ||
|
|
b529d0f44f | ||
|
|
c958b6496a | ||
|
|
858db2645d | ||
|
|
06380e5fe1 | ||
|
|
deaaa992bd | ||
|
|
715f832b00 | ||
|
|
c8dd6744da | ||
|
|
b398021b2c |
No files matched your search
@@ -140,7 +140,7 @@ void SnapshotPayloadParseState::flushQueuedKeys() {
|
||||
auto &insertsInFlightTmp = insertsInFlight; // C++ GRRRRRRRRRRRRRRRR, we don't want to capute "this" because that's dangerous
|
||||
if (current_database < cserver.dbnum) {
|
||||
g_pserver->asyncworkqueue->AddWorkFunction([idb, vecqueuedKeys = std::move(this->vecqueuedKeys), vecqueuedKeysCb = std::move(this->vecqueuedKeysCb), vecqueuedVals = std::move(this->vecqueuedVals), vecqueuedValsCb = std::move(this->vecqueuedValsCb), &insertsInFlightTmp, pallocator = m_spallocator.release()]() mutable {
|
||||
g_pserver->db[idb]->bulkStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
|
||||
g_pserver->db[idb]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
|
||||
--insertsInFlightTmp;
|
||||
delete pallocator;
|
||||
});
|
||||
|
||||
+29
-1
@@ -35,6 +35,7 @@
|
||||
|
||||
#include <fcntl.h>
|
||||
#include <sys/stat.h>
|
||||
#include <sys/wait.h>
|
||||
#ifdef __linux__
|
||||
#include <sys/sysinfo.h>
|
||||
#endif
|
||||
@@ -2456,6 +2457,32 @@ static int isValidAOFfilename(char *val, const char **err) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int isValidS3Bucket(char *s3bucket, const char **err) {
|
||||
int status = EXIT_FAILURE;
|
||||
pid_t pid = fork();
|
||||
if (pid < 0)
|
||||
{
|
||||
*err = "couldn't fork to call aws cli";
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (pid == 0)
|
||||
{
|
||||
execlp("aws", "aws", "s3", "ls", s3bucket);
|
||||
exit(EXIT_FAILURE);
|
||||
}
|
||||
else
|
||||
{
|
||||
waitpid(pid, &status, 0);
|
||||
}
|
||||
|
||||
if (status != EXIT_SUCCESS) {
|
||||
*err = "could not access s3 bucket";
|
||||
return 0;
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
||||
/* Validate specified string is a valid proc-title-template */
|
||||
static int isValidProcTitleTemplate(char *val, const char **err) {
|
||||
if (!validateProcTitleTemplate(val)) {
|
||||
@@ -2779,7 +2806,7 @@ standardConfig configs[] = {
|
||||
createBoolConfig("appendonly", NULL, MODIFIABLE_CONFIG, g_pserver->aof_enabled, 0, NULL, updateAppendonly),
|
||||
createBoolConfig("cluster-allow-reads-when-down", NULL, MODIFIABLE_CONFIG, g_pserver->cluster_allow_reads_when_down, 0, NULL, NULL),
|
||||
createBoolConfig("delete-on-evict", NULL, MODIFIABLE_CONFIG, cserver.delete_on_evict, 0, NULL, NULL),
|
||||
createBoolConfig("use-fork", NULL, IMMUTABLE_CONFIG, cserver.fForkBgSave, 0, NULL, NULL),
|
||||
createBoolConfig("use-fork", NULL, IMMUTABLE_CONFIG, cserver.fForkBgSave, 1, NULL, NULL),
|
||||
createBoolConfig("io-threads-do-reads", NULL, IMMUTABLE_CONFIG, fDummy, 0, NULL, NULL),
|
||||
createBoolConfig("time-thread-priority", NULL, IMMUTABLE_CONFIG, cserver.time_thread_priority, 0, NULL, NULL),
|
||||
createBoolConfig("prefetch-enabled", NULL, MODIFIABLE_CONFIG, g_pserver->prefetch_enabled, 1, NULL, NULL),
|
||||
@@ -2803,6 +2830,7 @@ standardConfig configs[] = {
|
||||
createStringConfig("cluster-announce-ip", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->cluster_announce_ip, NULL, NULL, NULL),
|
||||
createStringConfig("syslog-ident", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->syslog_ident, "redis", NULL, NULL),
|
||||
createStringConfig("dbfilename", NULL, MODIFIABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->rdb_filename, CONFIG_DEFAULT_RDB_FILENAME, isValidDBfilename, NULL),
|
||||
createStringConfig("db-s3-object", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->rdb_s3bucketpath, NULL, isValidS3Bucket, NULL),
|
||||
createStringConfig("appendfilename", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->aof_filename, "appendonly.aof", isValidAOFfilename, NULL),
|
||||
createStringConfig("server_cpulist", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->server_cpulist, NULL, NULL, NULL),
|
||||
createStringConfig("bio_cpulist", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->bio_cpulist, NULL, NULL, NULL),
|
||||
|
||||
+28
-9
@@ -1931,6 +1931,7 @@ void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e)
|
||||
* is associated with this key (i.e. the key is non volatile) */
|
||||
expireEntry *redisDbPersistentDataSnapshot::getExpire(const char *key) {
|
||||
/* No expire? return ASAP */
|
||||
std::unique_lock<fastlock> ul(g_expireLock);
|
||||
if (expireSize() == 0)
|
||||
return nullptr;
|
||||
|
||||
@@ -2607,12 +2608,15 @@ 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 moduleLoadCallback(const char * rgchKey, size_t, void *data) {
|
||||
redisObjectStack keyobj;
|
||||
initStaticStringObject(keyobj, const_cast<char *>(rgchKey));
|
||||
moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", &keyobj, *(int *)data);
|
||||
}
|
||||
|
||||
void moduleClusterLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
|
||||
clusterStorageLoadCallback(rgchKey, cchKey, data);
|
||||
moduleLoadCallback(rgchKey, cchKey, data);
|
||||
}
|
||||
|
||||
void redisDb::initialize(int id)
|
||||
@@ -2633,7 +2637,7 @@ void redisDb::storageProviderInitialize()
|
||||
{
|
||||
if (g_pserver->m_pstorageFactory != nullptr)
|
||||
{
|
||||
IStorageFactory::key_load_iterator itr = moduleLoadCallback;
|
||||
IStorageFactory::key_load_iterator itr = g_pserver->cluster_enabled ? moduleClusterLoadCallback : moduleLoadCallback;
|
||||
this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, &id));
|
||||
}
|
||||
}
|
||||
@@ -2724,8 +2728,11 @@ void redisDbPersistentData::clear(void(callback)(void*))
|
||||
m_cnewKeysPending = 0;
|
||||
m_fAllChanged++;
|
||||
}
|
||||
{
|
||||
std::unique_lock<fastlock> ul(g_expireLock);
|
||||
delete m_setexpire;
|
||||
m_setexpire = new (MALLOC_LOCAL) expireset();
|
||||
}
|
||||
if (m_spstorage != nullptr)
|
||||
m_spstorage->clear(callback);
|
||||
dictEmpty(m_pdictTombstone,callback);
|
||||
@@ -2858,7 +2865,8 @@ LNotFound:
|
||||
{
|
||||
dictAdd(m_pdict, sdsNewKey, o);
|
||||
o->SetFExpires(spexpire != nullptr);
|
||||
|
||||
|
||||
std::unique_lock<fastlock> ul(g_expireLock);
|
||||
if (spexpire != nullptr)
|
||||
{
|
||||
auto itr = m_setexpire->find(sdsKey);
|
||||
@@ -2881,6 +2889,7 @@ LNotFound:
|
||||
if (*pde != nullptr && dictGetVal(*pde) != nullptr)
|
||||
{
|
||||
robj *o = (robj*)dictGetVal(*pde);
|
||||
std::unique_lock<fastlock> ul(g_expireLock);
|
||||
serverAssert(o->FExpires() == (m_setexpire->find(sdsKey) != m_setexpire->end()));
|
||||
}
|
||||
}
|
||||
@@ -3002,8 +3011,16 @@ void redisDbPersistentData::processChangesAsync(std::atomic<int> &pendingJobs)
|
||||
});
|
||||
}
|
||||
|
||||
void redisDbPersistentData::bulkStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem)
|
||||
/* This function is to bulk insert directly to storage provider bypassing in memory, assumes rgKeys and rgVals are not sds strings */
|
||||
void redisDbPersistentData::bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem)
|
||||
{
|
||||
if (g_pserver->cluster_enabled) {
|
||||
aeAcquireLock();
|
||||
for (size_t i = 0; i < celem; i++) {
|
||||
slotToKeyUpdateKeyCore(rgKeys[i], rgcbKeys[i], 1);
|
||||
}
|
||||
aeReleaseLock();
|
||||
}
|
||||
m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, celem);
|
||||
}
|
||||
|
||||
@@ -3245,6 +3262,7 @@ std::unique_ptr<expireEntry> deserializeExpire(sds key, const char *str, size_t
|
||||
|
||||
sds serializeStoredObjectAndExpire(redisDbPersistentData *db, const char *key, robj_roptr o)
|
||||
{
|
||||
std::unique_lock<fastlock> ul(g_expireLock);
|
||||
auto itrExpire = db->setexpire()->find(key);
|
||||
const expireEntry *pexpire = nullptr;
|
||||
if (itrExpire != db->setexpire()->end())
|
||||
@@ -3369,6 +3387,7 @@ void redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
|
||||
dictAdd(m_pdict, sharedKey, o);
|
||||
o->SetFExpires(spexpire != nullptr);
|
||||
|
||||
std::unique_lock<fastlock> ul(g_expireLock);
|
||||
if (spexpire != nullptr)
|
||||
{
|
||||
auto itr = m_setexpire->find(sharedKey);
|
||||
|
||||
@@ -481,6 +481,7 @@ void expireSlaveKeys(void) {
|
||||
if (slaveKeysWithExpire == NULL ||
|
||||
dictSize(slaveKeysWithExpire) == 0) return;
|
||||
|
||||
std::unique_lock<fastlock> ul(g_expireLock);
|
||||
int cycles = 0, noexpire = 0;
|
||||
mstime_t start = mstime();
|
||||
while(1) {
|
||||
|
||||
@@ -1668,6 +1668,7 @@ robj *deserializeStoredObjectCore(const void *data, size_t cb)
|
||||
robj *deserializeStoredObject(const redisDbPersistentData *db, const char *key, const void *data, size_t cb)
|
||||
{
|
||||
robj *o = deserializeStoredObjectCore(data, cb);
|
||||
std::unique_lock<fastlock> ul(g_expireLock);
|
||||
o->SetFExpires(db->setexpire()->exists(key));
|
||||
return o;
|
||||
}
|
||||
|
||||
+120
-47
@@ -1334,7 +1334,7 @@ int rdbSaveRio(rio *rdb, const redisDbPersistentDataSnapshot **rgpdb, int *error
|
||||
if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_BEFORE_RDB) == -1) goto werr;
|
||||
|
||||
for (j = 0; j < cserver.dbnum; j++) {
|
||||
const redisDbPersistentDataSnapshot *db = rgpdb[j];
|
||||
const redisDbPersistentDataSnapshot *db = rgpdb != nullptr ? rgpdb[j] : g_pserver->db[j];
|
||||
if (db->size() == 0) continue;
|
||||
|
||||
/* Write the SELECT DB opcode */
|
||||
@@ -1715,7 +1715,7 @@ void getTempFileName(char tmpfile[], int tmpfileNum) {
|
||||
char tmpfileNumString[214];
|
||||
|
||||
/* Generate temp rdb file name using aync-signal safe functions. */
|
||||
int pid_len = ll2string(pid, sizeof(pid), getpid());
|
||||
int pid_len = ll2string(pid, sizeof(pid), g_pserver->in_fork_child ? getppid() : getpid());
|
||||
int tmpfileNum_len = ll2string(tmpfileNumString, sizeof(tmpfileNumString), tmpfileNum);
|
||||
strcpy(tmpfile, "temp-");
|
||||
strncpy(tmpfile+5, pid, pid_len);
|
||||
@@ -3832,57 +3832,130 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
|
||||
}
|
||||
|
||||
/* Create the child process. */
|
||||
openChildInfoPipe();
|
||||
if (cserver.fForkBgSave) {
|
||||
pid_t childpid;
|
||||
if ((childpid = redisFork(CHILD_TYPE_RDB)) == 0) {
|
||||
/* Child */
|
||||
int retval, dummy;
|
||||
rio rdb;
|
||||
|
||||
for (int idb = 0; idb < cserver.dbnum; ++idb)
|
||||
args->rgpdb[idb] = g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false /*fOptional*/);
|
||||
rioInitWithFd(&rdb,args->rdb_pipe_write);
|
||||
|
||||
g_pserver->rdbThreadVars.tmpfileNum++;
|
||||
g_pserver->rdbThreadVars.fRdbThreadCancel = false;
|
||||
pthread_attr_t tattr;
|
||||
pthread_attr_init(&tattr);
|
||||
pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
|
||||
if (pthread_create(&child, &tattr, rdbSaveToSlavesSocketsThread, args)) {
|
||||
pthread_attr_destroy(&tattr);
|
||||
serverLog(LL_WARNING,"Can't save in background: fork: %s",
|
||||
strerror(errno));
|
||||
redisSetProcTitle("keydb-rdb-to-slaves");
|
||||
redisSetCpuAffinity(g_pserver->bgsave_cpulist);
|
||||
|
||||
/* Undo the state change. The caller will perform cleanup on
|
||||
* all the slaves in BGSAVE_START state, but an early call to
|
||||
* replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
|
||||
listRewind(g_pserver->slaves,&li);
|
||||
while((ln = listNext(&li))) {
|
||||
client *replica = (client*)ln->value;
|
||||
if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
|
||||
replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
|
||||
retval = rdbSaveRioWithEOFMark(&rdb,nullptr,nullptr,rsi);
|
||||
if (retval == C_OK && rioFlush(&rdb) == 0)
|
||||
retval = C_ERR;
|
||||
|
||||
if (retval == C_OK) {
|
||||
sendChildCowInfo(CHILD_INFO_TYPE_RDB_COW_SIZE, "RDB");
|
||||
}
|
||||
}
|
||||
close(args->rdb_pipe_write);
|
||||
close(g_pserver->rdb_pipe_read);
|
||||
zfree(g_pserver->rdb_pipe_conns);
|
||||
close(args->safe_to_exit_pipe);
|
||||
g_pserver->rdb_pipe_conns = NULL;
|
||||
g_pserver->rdb_pipe_numconns = 0;
|
||||
g_pserver->rdb_pipe_numconns_writing = 0;
|
||||
args->rsi.~rdbSaveInfo();
|
||||
zfree(args);
|
||||
closeChildInfoPipe();
|
||||
return C_ERR;
|
||||
}
|
||||
pthread_attr_destroy(&tattr);
|
||||
g_pserver->child_type = CHILD_TYPE_RDB;
|
||||
|
||||
serverLog(LL_NOTICE,"Background RDB transfer started");
|
||||
g_pserver->rdb_save_time_start = time(NULL);
|
||||
serverAssert(!g_pserver->rdbThreadVars.fRdbThreadActive);
|
||||
g_pserver->rdbThreadVars.rdb_child_thread = child;
|
||||
g_pserver->rdbThreadVars.fRdbThreadActive = true;
|
||||
g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
|
||||
aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
|
||||
if (aeCreateFileEvent(serverTL->el, g_pserver->rdb_pipe_read, AE_READABLE, rdbPipeReadHandler,NULL) == AE_ERR) {
|
||||
serverPanic("Unrecoverable error creating server.rdb_pipe_read file event.");
|
||||
rioFreeFd(&rdb);
|
||||
/* wake up the reader, tell it we're done. */
|
||||
close(args->rdb_pipe_write);
|
||||
close(g_pserver->rdb_child_exit_pipe); /* close write end so that we can detect the close on the parent. */
|
||||
/* hold exit until the parent tells us it's safe. we're not expecting
|
||||
* to read anything, just get the error when the pipe is closed. */
|
||||
dummy = read(args->safe_to_exit_pipe, pipefds, 1);
|
||||
UNUSED(dummy);
|
||||
exitFromChild((retval == C_OK) ? 0 : 1);
|
||||
} else {
|
||||
/* Parent */
|
||||
close(args->safe_to_exit_pipe);
|
||||
if (childpid == -1) {
|
||||
serverLog(LL_WARNING,"Can't save in background: fork: %s",
|
||||
strerror(errno));
|
||||
|
||||
/* Undo the state change. The caller will perform cleanup on
|
||||
* all the slaves in BGSAVE_START state, but an early call to
|
||||
* replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
|
||||
listRewind(g_pserver->slaves,&li);
|
||||
while((ln = listNext(&li))) {
|
||||
client *replica = (client*)ln->value;
|
||||
if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
|
||||
replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
|
||||
}
|
||||
}
|
||||
close(args->rdb_pipe_write);
|
||||
close(g_pserver->rdb_pipe_read);
|
||||
zfree(g_pserver->rdb_pipe_conns);
|
||||
g_pserver->rdb_pipe_conns = NULL;
|
||||
g_pserver->rdb_pipe_numconns = 0;
|
||||
g_pserver->rdb_pipe_numconns_writing = 0;
|
||||
args->rsi.~rdbSaveInfo();
|
||||
zfree(args);
|
||||
} else {
|
||||
serverLog(LL_NOTICE,"Background RDB transfer started by pid %ld",
|
||||
(long)childpid);
|
||||
g_pserver->rdb_save_time_start = time(NULL);
|
||||
g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
|
||||
g_pserver->rdbThreadVars.fRdbThreadActive = true;
|
||||
updateDictResizePolicy();
|
||||
close(args->rdb_pipe_write); /* close write in parent so that it can detect the close on the child. */
|
||||
aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
|
||||
if (aeCreateFileEvent(serverTL->el, g_pserver->rdb_pipe_read, AE_READABLE, rdbPipeReadHandler, nullptr) == AE_ERR) {
|
||||
serverPanic("Unrecoverable error creating g_pserver->rdb_pipe_read file event.");
|
||||
}
|
||||
});
|
||||
}
|
||||
return (childpid == -1) ? C_ERR : C_OK;
|
||||
}
|
||||
});
|
||||
}
|
||||
else {
|
||||
openChildInfoPipe();
|
||||
|
||||
for (int idb = 0; idb < cserver.dbnum; ++idb)
|
||||
args->rgpdb[idb] = g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false /*fOptional*/);
|
||||
|
||||
g_pserver->rdbThreadVars.tmpfileNum++;
|
||||
g_pserver->rdbThreadVars.fRdbThreadCancel = false;
|
||||
pthread_attr_t tattr;
|
||||
pthread_attr_init(&tattr);
|
||||
pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
|
||||
if (pthread_create(&child, &tattr, rdbSaveToSlavesSocketsThread, args)) {
|
||||
pthread_attr_destroy(&tattr);
|
||||
serverLog(LL_WARNING,"Can't save in background: fork: %s",
|
||||
strerror(errno));
|
||||
|
||||
/* Undo the state change. The caller will perform cleanup on
|
||||
* all the slaves in BGSAVE_START state, but an early call to
|
||||
* replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
|
||||
listRewind(g_pserver->slaves,&li);
|
||||
while((ln = listNext(&li))) {
|
||||
client *replica = (client*)ln->value;
|
||||
if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
|
||||
replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
|
||||
}
|
||||
}
|
||||
close(args->rdb_pipe_write);
|
||||
close(g_pserver->rdb_pipe_read);
|
||||
zfree(g_pserver->rdb_pipe_conns);
|
||||
close(args->safe_to_exit_pipe);
|
||||
g_pserver->rdb_pipe_conns = NULL;
|
||||
g_pserver->rdb_pipe_numconns = 0;
|
||||
g_pserver->rdb_pipe_numconns_writing = 0;
|
||||
args->rsi.~rdbSaveInfo();
|
||||
zfree(args);
|
||||
closeChildInfoPipe();
|
||||
return C_ERR;
|
||||
}
|
||||
pthread_attr_destroy(&tattr);
|
||||
g_pserver->child_type = CHILD_TYPE_RDB;
|
||||
|
||||
serverLog(LL_NOTICE,"Background RDB transfer started");
|
||||
g_pserver->rdb_save_time_start = time(NULL);
|
||||
serverAssert(!g_pserver->rdbThreadVars.fRdbThreadActive);
|
||||
g_pserver->rdbThreadVars.rdb_child_thread = child;
|
||||
g_pserver->rdbThreadVars.fRdbThreadActive = true;
|
||||
g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
|
||||
aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
|
||||
if (aeCreateFileEvent(serverTL->el, g_pserver->rdb_pipe_read, AE_READABLE, rdbPipeReadHandler, nullptr) == AE_ERR) {
|
||||
serverPanic("Unrecoverable error creating server.rdb_pipe_read file event.");
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
return C_OK; /* Unreached. */
|
||||
}
|
||||
|
||||
+5
-3
@@ -3797,9 +3797,11 @@ int connectWithMaster(redisMaster *mi) {
|
||||
connSetPrivateData(mi->repl_transfer_s, mi);
|
||||
if (connConnect(mi->repl_transfer_s, mi->masterhost, mi->masterport,
|
||||
NET_FIRST_BIND_ADDR, syncWithMaster) == C_ERR) {
|
||||
int sev = g_pserver->enable_multimaster ? LL_NOTICE : LL_WARNING; // with multimaster its not unheard of to intentiallionall have downed masters
|
||||
serverLog(sev,"Unable to connect to MASTER: %s",
|
||||
connGetLastError(mi->repl_transfer_s));
|
||||
const char *err = "Unknown Error";
|
||||
if (mi->repl_transfer_s->last_errno != 0)
|
||||
err = connGetLastError(mi->repl_transfer_s);
|
||||
int sev = g_pserver->enable_multimaster ? LL_NOTICE : LL_WARNING; // with multimaster its not unheard of to intentionally have downed masters
|
||||
serverLog(sev, "Unable to connect to MASTER: %s", err);
|
||||
connClose(mi->repl_transfer_s);
|
||||
mi->repl_transfer_s = NULL;
|
||||
return C_ERR;
|
||||
|
||||
@@ -1264,6 +1264,9 @@ void scriptingInit(int setup) {
|
||||
|
||||
/* Finally set the table as 'redis' global var. */
|
||||
lua_setglobal(lua,"redis");
|
||||
/* Set table as 'keydb' global var as well */
|
||||
lua_getglobal(lua,"redis");
|
||||
lua_setglobal(lua,"keydb");
|
||||
|
||||
/* Replace math.random and math.randomseed with our implementations. */
|
||||
lua_getglobal(lua,"math");
|
||||
|
||||
+1
-1
@@ -5172,7 +5172,7 @@ int prepareForShutdown(int flags) {
|
||||
* to unlink file actully) in background thread.
|
||||
* The temp rdb file fd may won't be closed when redis exits quickly,
|
||||
* but OS will close this fd when process exits. */
|
||||
rdbRemoveTempFile(g_pserver->child_pid, 0);
|
||||
rdbRemoveTempFile(g_pserver->rdbThreadVars.tmpfileNum, 0);
|
||||
}
|
||||
|
||||
/* Kill module child if there is one. */
|
||||
|
||||
+2
-2
@@ -1209,7 +1209,7 @@ public:
|
||||
bool FSnapshot() const { return m_spdbSnapshotHOLDER != nullptr; }
|
||||
|
||||
std::unique_ptr<const StorageCache> CloneStorageCache() { return std::unique_ptr<const StorageCache>(m_spstorage->clone()); }
|
||||
void bulkStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
|
||||
void bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
|
||||
|
||||
dict_iter find_cached_threadsafe(const char *key) const;
|
||||
|
||||
@@ -1370,7 +1370,7 @@ struct redisDb : public redisDbPersistentDataSnapshot
|
||||
using redisDbPersistentData::FRehashing;
|
||||
using redisDbPersistentData::FTrackingChanges;
|
||||
using redisDbPersistentData::CloneStorageCache;
|
||||
using redisDbPersistentData::bulkStorageInsert;
|
||||
using redisDbPersistentData::bulkDirectStorageInsert;
|
||||
|
||||
public:
|
||||
const redisDbPersistentDataSnapshot *createSnapshot(uint64_t mvccCheckpoint, bool fOptional) {
|
||||
|
||||
@@ -162,15 +162,16 @@ start_server {tags {"introspection"}} {
|
||||
aof_rewrite_cpulist
|
||||
time-thread-priority
|
||||
bgsave_cpulist
|
||||
storage-cache-mode
|
||||
storage-provider-options
|
||||
use-fork
|
||||
storage-cache-mode
|
||||
storage-provider-options
|
||||
use-fork
|
||||
multi-master
|
||||
active-replica
|
||||
bind
|
||||
set-proc-title
|
||||
repl-backlog-disk-reserve
|
||||
tls-allowlist
|
||||
db-s3-object
|
||||
}
|
||||
|
||||
if {!$::tls} {
|
||||
|
||||
@@ -43,10 +43,15 @@ start_server {tags {"scripting"}} {
|
||||
r eval {return redis.call('get',KEYS[1])} 1 mykey
|
||||
} {myval}
|
||||
|
||||
test { EVAL - keys command works? } {
|
||||
test {EVAL - keys command works? } {
|
||||
r eval {return redis.call('keys', 'test')} 0
|
||||
}
|
||||
|
||||
test {EVAL - KeyDB global works } {
|
||||
r eval {return keydb.call('get', KEYS[1])} 1 mykey
|
||||
assert_equal [r eval {return redis.call('get',KEYS[1])} 1 mykey] [r eval {return keydb.call('get', KEYS[1])} 1 mykey]
|
||||
}
|
||||
|
||||
test {EVALSHA - Can we call a SHA1 if already defined?} {
|
||||
r evalsha fd758d1589d044dd850a6f05d52f2eefd27f033f 1 mykey
|
||||
} {myval}
|
||||
|
||||
@@ -26,7 +26,11 @@ then
|
||||
while [ $((PORT < ENDPORT)) != "0" ]; do
|
||||
PORT=$((PORT+1))
|
||||
echo "Starting $PORT"
|
||||
$BIN_PATH/keydb-server --port $PORT --protected-mode $PROTECTED_MODE --cluster-enabled yes --cluster-config-file nodes-${PORT}.conf --cluster-node-timeout $TIMEOUT --appendonly yes --appendfilename appendonly-${PORT}.aof --dbfilename dump-${PORT}.rdb --logfile ${PORT}.log --daemonize yes ${ADDITIONAL_OPTIONS}
|
||||
if [ "$2" == "flash" ]
|
||||
then
|
||||
ADDITIONAL_OPTIONS="--save \"\" \"\" \"\" --semi-ordered-set-bucket-size 8 --client-output-buffer-limit replica 1 1 0 --maxmemory 100000000 --storage-provider flash ./$PORT.flash"
|
||||
fi
|
||||
$BIN_PATH/keydb-server --server-threads 4 --port $PORT --protected-mode $PROTECTED_MODE --cluster-enabled yes --cluster-config-file nodes-${PORT}.conf --cluster-node-timeout $TIMEOUT --dbfilename dump-${PORT}.rdb --logfile ${PORT}.log --daemonize yes ${ADDITIONAL_OPTIONS}
|
||||
done
|
||||
exit 0
|
||||
fi
|
||||
@@ -97,6 +101,8 @@ then
|
||||
rm -rf appendonly*.aof
|
||||
rm -rf dump*.rdb
|
||||
rm -rf nodes*.conf
|
||||
rm -rf *.flash
|
||||
rm -rf temp*.rdb
|
||||
exit 0
|
||||
fi
|
||||
|
||||
|
||||
Reference in new issue
Block a user