full data sync #8
243
src/rdbLoad.c
243
src/rdbLoad.c
@ -1,19 +1,25 @@
|
||||
#include "rdbLoad.h"
|
||||
#include "sds.h"
|
||||
#include "sdscompat.h"
|
||||
|
||||
#include <errno.h>
|
||||
#include <unistd.h>
|
||||
#include <sys/time.h>
|
||||
#include <unistd.h>
|
||||
#include <string.h>
|
||||
|
||||
int rdbLoadRioWithLoading(migrateObj *mobj) {
|
||||
int rmLoadRioWithLoading(migrateObj *mobj) {
|
||||
char buf[1024];
|
||||
int error;
|
||||
if (syncReadLine(mobj->source_cc->fd, buf, 9, mobj->timeout) == -1) {
|
||||
serverLog(LL_WARNING, "read version failed:%s,port:%d ", mobj->host,
|
||||
serverLog(LL_WARNING, "[rm]read version failed:%s,port:%d ", mobj->host,
|
||||
mobj->port);
|
||||
return C_ERR;
|
||||
}
|
||||
buf[9] = '\0';
|
||||
serverLog(LL_NOTICE, "[rm] version=%s", buf);
|
||||
if (memcmp(buf, "REDIS", 5) != 0) {
|
||||
serverLog(LL_WARNING, "Wrong signature trying to load DB from file");
|
||||
serverLog(LL_WARNING, "[rm] Wrong signature trying to load DB from file");
|
||||
errno = EINVAL;
|
||||
return C_ERR;
|
||||
}
|
||||
@ -25,25 +31,39 @@ int rdbLoadRioWithLoading(migrateObj *mobj) {
|
||||
errno = EINVAL;
|
||||
return C_ERR;
|
||||
}
|
||||
serverLog(LL_NOTICE, "buf=%s", buf);
|
||||
long long lru_idle = -1, lfu_freq = -1, expiretime = -1, now = mstime();
|
||||
|
||||
// syncReadLine(mobj->source_cc->fd, buf, 4, mobj->timeout);
|
||||
// buf[4] = '\0';
|
||||
// serverLog(LL_NOTICE, "buf:%s", buf);
|
||||
// type = atoi(buf);
|
||||
// serverLog(LL_NOTICE, "type:%d", type);
|
||||
char type_buf[1];
|
||||
if (read(mobj->source_cc->fd, &type_buf, 1) == 0)
|
||||
{
|
||||
serverLog(LL_WARNING, "read mark failed");
|
||||
return C_ERR;
|
||||
}
|
||||
|
||||
while (1) {
|
||||
sds key;
|
||||
robj *val;
|
||||
if ((type = rm_rdbLoadType(mobj)) == -1) {
|
||||
if (read(mobj->source_cc->fd, &type_buf, 1) == 0) {
|
||||
serverLog(LL_WARNING, "read type failed");
|
||||
return C_ERR;
|
||||
}
|
||||
int type = type_buf[0] & 0xff;
|
||||
serverLog(LL_NOTICE, "type: %d", type);
|
||||
|
||||
if (type == RDB_OPCODE_EXPIRETIME) {
|
||||
expiretime = rm_rdbLoadTime(mobj);
|
||||
expiretime = rmLoadTime(mobj);
|
||||
if (expiretime == -1) {
|
||||
return C_ERR;
|
||||
}
|
||||
expiretime *= 1000;
|
||||
continue;
|
||||
} else if (type == RDB_OPCODE_EXPIRETIME_MS) {
|
||||
expiretime = rm_rdbLoadMillisecondTime(mobj, rdbver);
|
||||
expiretime = rmLoadMillisecondTime(mobj, rdbver);
|
||||
continue;
|
||||
} else if (type == RDB_OPCODE_FREQ) {
|
||||
uint8_t byte;
|
||||
@ -53,56 +73,235 @@ int rdbLoadRioWithLoading(migrateObj *mobj) {
|
||||
continue;
|
||||
} else if (type == RDB_OPCODE_IDLE) {
|
||||
uint64_t qword;
|
||||
if ((qword = rm_rdbLoadLen(mobj, NULL)) == RDB_LENERR)
|
||||
if ((qword = rmLoadLen(mobj, NULL)) == RDB_LENERR)
|
||||
return C_ERR;
|
||||
lru_idle = qword;
|
||||
continue;
|
||||
} else if (type == RDB_OPCODE_EOF) {
|
||||
serverLog(LL_NOTICE, "rdb file parse end..");
|
||||
break;
|
||||
} else if (type == RDB_OPCODE_SELECTDB) {
|
||||
if ((dbid = rm_rdbLoadLen(mobj, NULL)) == RDB_LENERR) return C_ERR;
|
||||
if ((dbid = rmLoadLen(mobj, NULL)) == RDB_LENERR) return C_ERR;
|
||||
continue;
|
||||
} else if (type == RDB_OPCODE_RESIZEDB) {
|
||||
uint64_t db_size, expires_size;
|
||||
if ((db_size = rm_rdbLoadLen(mobj, NULL)) == RDB_LENERR) return C_ERR;
|
||||
if ((expires_size = rm_rdbLoadLen(mobj, NULL)) == RDB_LENERR) return C_ERR;
|
||||
|
||||
if ((db_size = rmLoadLen(mobj, NULL)) == RDB_LENERR) return C_ERR;
|
||||
if ((expires_size = rmLoadLen(mobj, NULL)) == RDB_LENERR) return C_ERR;
|
||||
continue;
|
||||
} else if (type == RDB_OPCODE_AUX) {
|
||||
robj *auxkey, *auxval;
|
||||
sds *auxkey, *auxval;
|
||||
if ((auxkey = rmLoadStringObject(mobj)) == NULL) {
|
||||
serverLog(LL_WARNING, "auxkey is null");
|
||||
return NULL;
|
||||
}
|
||||
if ((auxval = rmLoadStringObject(mobj)) == NULL) {
|
||||
serverLog(LL_WARNING, "auxval is null");
|
||||
return NULL;
|
||||
}
|
||||
if (auxkey[0] == '%') {
|
||||
serverLog(LL_NOTICE, "RDB '%s': %s", auxkey, auxval);
|
||||
} else if (!strcasecmp(auxkey, "repl-id")) {
|
||||
memcmp(mobj->psync_replid, auxval, CONFIG_RUN_ID_SIZE + 1);
|
||||
serverLog(LL_NOTICE, "repl-id: %s", auxval);
|
||||
} else if (!strcasecmp(auxkey, "repl-offset")) {
|
||||
serverLog(LL_NOTICE, "repl-offset: %s", auxval);
|
||||
} else if (!strcasecmp(auxkey, "lua") || !strcasecmp(auxkey, "aof-base")
|
||||
|| !strcasecmp(auxkey, "redis-bits") || !strcasecmp(auxkey, "aof-preamble")
|
||||
|| !strcasecmp(auxkey, "repl-stream-db") || !strcasecmp(auxkey, "redis-ver")
|
||||
|| !strcasecmp(auxkey, "ctime")) {
|
||||
serverLog(LL_NOTICE, "RDB '%s': %s", auxkey, auxval);
|
||||
} else if (!strcasecmp(auxkey, "used-mem")) {
|
||||
long long usedmem = strtoll(auxval, NULL, 10);
|
||||
serverLog(LL_NOTICE, "RDB memory usage when created %.2f Mb", (double)usedmem / (1024 * 1024));
|
||||
} else {
|
||||
serverLog(LL_DEBUG, "Unrecognized RDB AUX field: '%s'", auxkey);
|
||||
}
|
||||
continue;
|
||||
|
||||
} else if (type == RDB_OPCODE_FUNCTION || type == RDB_OPCODE_FUNCTION2) {
|
||||
continue;
|
||||
}
|
||||
if ((key = rmGenericLoadStringObject(mobj, RDB_LOAD_SDS, NULL)) == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
val = rmLoadObject(type, mobj, key, &error);
|
||||
|
||||
if (val == NULL) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void *rm_rdbGenericLoadStringObject(migrateObj *mobj, int flags, size_t *lenptr) {
|
||||
robj *rmLoadObject(int rdbtype, migrateObj *mobj, sds key, int *error) {
|
||||
int dbid = 0;
|
||||
robj *o = NULL, *ele, *dec;
|
||||
uint64_t len;
|
||||
unsigned int i;
|
||||
|
||||
return o;
|
||||
}
|
||||
|
||||
sds rmLoadStringObject(migrateObj *mobj) {
|
||||
return rmGenericLoadStringObject(mobj, RDB_LOAD_NONE, NULL);
|
||||
}
|
||||
|
||||
sds rmGenericLoadStringObject(migrateObj *mobj, int flags, size_t *lenptr) {
|
||||
int encode = flags & RDB_LOAD_ENC;
|
||||
int plain = flags & RDB_LOAD_PLAIN;
|
||||
int sds = flags & RDB_LOAD_SDS;
|
||||
int isencoded;
|
||||
unsigned long long len;
|
||||
|
||||
len = rmLoadLen(mobj, &isencoded);
|
||||
serverLog(LL_NOTICE, "len=%d, isencoded=%d", len, isencoded);
|
||||
if (len == RDB_LENERR) return NULL;
|
||||
if (isencoded) {
|
||||
switch (len) {
|
||||
case RDB_ENC_INT8:
|
||||
case RDB_ENC_INT16:
|
||||
case RDB_ENC_INT32:
|
||||
return rmLoadIntegerObject(mobj, len, flags, lenptr);
|
||||
case RDB_ENC_LZF:
|
||||
return rmLoadLzfStringObject(mobj, flags, lenptr);
|
||||
default:
|
||||
serverLog(LL_WARNING, "Unknown RDB string encoding type %llu", len);
|
||||
return NULL;
|
||||
}
|
||||
}
|
||||
if (plain || sds) {
|
||||
void *buf = sds_malloc(len);
|
||||
if (!buf) {
|
||||
serverLog(LL_WARNING, "rdbGenericLoadStringObject failed allocating %llu bytes", len);
|
||||
return NULL;
|
||||
}
|
||||
if (lenptr) *lenptr = len;
|
||||
if (len && read(mobj->source_cc->fd, buf, len) == 0) {
|
||||
if (plain)
|
||||
zfree(buf);
|
||||
else
|
||||
sdsfree(buf);
|
||||
return NULL;
|
||||
}
|
||||
return hi_sdsnewlen(buf, len);
|
||||
} else {
|
||||
char buf[len];
|
||||
if (!buf) {
|
||||
serverLog(LL_WARNING, "rdbGenericLoadStringObject failed allocating %llu bytes", len);
|
||||
return NULL;
|
||||
}
|
||||
if (len && read(mobj->source_cc->fd, buf, len) == 0) {
|
||||
return NULL;
|
||||
}
|
||||
buf[len] = '\0';
|
||||
return hi_sdsnewlen(buf, len);
|
||||
}
|
||||
}
|
||||
|
||||
uint64_t rm_rdbLoadLen(migrateObj *mobj, int *isencoded) {
|
||||
void *rmLoadIntegerObject(migrateObj *mobj, int enctype, int flags, size_t *lenptr) {
|
||||
int plain = flags & RDB_LOAD_PLAIN;
|
||||
int sds = flags & RDB_LOAD_SDS;
|
||||
int encode = flags & RDB_LOAD_ENC;
|
||||
unsigned char enc[4];
|
||||
long long val;
|
||||
if (enctype == RDB_ENC_INT8) {
|
||||
if (read(mobj->source_cc->fd, enc, 1) == 0) return NULL;
|
||||
val = (signed char)enc[0];
|
||||
} else if (enctype == RDB_ENC_INT16) {
|
||||
uint16_t v;
|
||||
if (read(mobj->source_cc->fd, enc, 2) == 0) return NULL;
|
||||
v = ((uint32_t)enc[0]) | ((uint32_t)enc[1] << 8);
|
||||
val = (int16_t)v;
|
||||
} else if (enctype == RDB_ENC_INT32) {
|
||||
uint32_t v;
|
||||
if (read(mobj->source_cc->fd, enc, 4) == 0) return NULL;
|
||||
v = ((uint32_t)enc[0]) | ((uint32_t)enc[1] << 8) | ((uint32_t)enc[2] << 16) | ((uint32_t)enc[3] << 24);
|
||||
val = (int32_t)v;
|
||||
} else {
|
||||
// rdbReportCorruptRDB("Unknown RDB integer encoding type %d", enctype);
|
||||
return NULL; /* Never reached. */
|
||||
}
|
||||
serverLog(LL_NOTICE, "plain=%d, encode=%d", plain, encode);
|
||||
if (plain || sds) {
|
||||
char buf[LONG_STR_SIZE], *p;
|
||||
int len = ll2string(buf, sizeof(buf), val);
|
||||
if (lenptr) *lenptr = len;
|
||||
p = plain ? zmalloc(len) : sdsnewlen(SDS_NOINIT, len);
|
||||
memcpy(p, buf, len);
|
||||
return p;
|
||||
} else if (encode) {
|
||||
return createStringObjectFromLongLongForValue(val);
|
||||
} else {
|
||||
return sdsfromlonglong(val);
|
||||
}
|
||||
}
|
||||
|
||||
void *rmLoadLzfStringObject(migrateObj *mobj, int flags, size_t *lenptr) {
|
||||
int plain = flags & RDB_LOAD_PLAIN;
|
||||
int sds = flags & RDB_LOAD_SDS;
|
||||
uint64_t len, clen;
|
||||
unsigned char *c = NULL;
|
||||
char *val = NULL;
|
||||
if ((clen = rmLoadLen(mobj, NULL)) == RDB_LENERR) return NULL;
|
||||
if ((len = rmLoadLen(mobj, NULL)) == RDB_LENERR) return NULL;
|
||||
if ((c = hi_malloc(clen)) == NULL) {
|
||||
serverLog(LL_WARNING, "rdbLoadLzfStringObject failed allocating %llu bytes", (unsigned long long)clen);
|
||||
goto err;
|
||||
}
|
||||
if (plain) {
|
||||
val = hi_malloc(len);
|
||||
} else {
|
||||
val = hi_sdsnewlen(SDS_NOINIT, len);
|
||||
}
|
||||
if (!val) {
|
||||
serverLog(LL_WARNING, "rdbLoadLzfStringObject failed allocating %llu bytes", (unsigned long long)len);
|
||||
goto err;
|
||||
}
|
||||
if (lenptr) *lenptr = len;
|
||||
if (read(mobj->source_cc->fd, c, len) == 0) return NULL;
|
||||
if (lzf_decompress(c, clen, val, len) != len) {
|
||||
serverLog(LL_WARNING, "Invalid LZF compressed string");
|
||||
goto err;
|
||||
}
|
||||
zfree(c);
|
||||
|
||||
if (plain || sds) {
|
||||
return val;
|
||||
} else {
|
||||
return createObject(OBJ_STRING, val);
|
||||
}
|
||||
err:
|
||||
zfree(c);
|
||||
if (plain)
|
||||
zfree(val);
|
||||
else
|
||||
sdsfree(val);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
uint64_t rmLoadLen(migrateObj *mobj, int *isencoded) {
|
||||
uint64_t len;
|
||||
|
||||
if (rdbLoadLenByRef(mobj, isencoded, &len) == -1)
|
||||
if (rmLoadLenByRef(mobj, isencoded, &len) == -1)
|
||||
return RDB_LENERR;
|
||||
return len;
|
||||
}
|
||||
|
||||
int rm_rdbLoadType(migrateObj *mobj) {
|
||||
int rmLoadType(migrateObj *mobj) {
|
||||
unsigned char type;
|
||||
if (read(mobj->source_cc->fd, &type, 1) == 0) {
|
||||
if (rmRead(mobj, &type, 1) == 0) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
return type;
|
||||
}
|
||||
|
||||
int rm_rdbLoadLenByRef(migrateObj *mobj, int *isencoded, uint64_t *lenptr) {
|
||||
int rmRead(migrateObj *mobj, void *buf, size_t len) {
|
||||
if (read(mobj->source_cc->fd, &buf, len) == 0) {
|
||||
return 0;
|
||||
}
|
||||
buf = (char *)buf + len;
|
||||
return 1;
|
||||
}
|
||||
|
||||
int rmLoadLenByRef(migrateObj *mobj, int *isencoded, uint64_t *lenptr) {
|
||||
unsigned char buf[2];
|
||||
int type;
|
||||
if (isencoded)
|
||||
@ -138,7 +337,7 @@ int rm_rdbLoadLenByRef(migrateObj *mobj, int *isencoded, uint64_t *lenptr) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
time_t rm_rdbLoadTime(migrateObj *mobj) {
|
||||
time_t rmLoadTime(migrateObj *mobj) {
|
||||
int32_t t32;
|
||||
if (read(mobj->source_cc->fd, &t32, 4) == 0) {
|
||||
return -1;
|
||||
@ -147,7 +346,7 @@ time_t rm_rdbLoadTime(migrateObj *mobj) {
|
||||
return (time_t)t32;
|
||||
}
|
||||
|
||||
long long rm_rdbLoadMillisecondTime(migrateObj *mobj, int rdbver) {
|
||||
long long rmLoadMillisecondTime(migrateObj *mobj, int rdbver) {
|
||||
int64_t t64;
|
||||
if (read(mobj->source_cc->fd, &t64, 8) == 0) {
|
||||
return LLONG_MAX;
|
||||
|
||||
@ -27,22 +27,48 @@
|
||||
#define RDB_LOAD_PLAIN (1 << 1)
|
||||
#define RDB_LOAD_SDS (1 << 2)
|
||||
|
||||
#define RDB_ENC_INT8 0 /* 8 bit signed integer */
|
||||
#define RDB_ENC_INT16 1 /* 16 bit signed integer */
|
||||
#define RDB_ENC_INT32 2 /* 32 bit signed integer */
|
||||
#define RDB_ENC_LZF 3 /* string compressed with FASTLZ */
|
||||
|
||||
#define LONG_STR_SIZE 21
|
||||
extern const char *SDS_NOINIT;
|
||||
#define OBJ_STRING 0 /* String object. */
|
||||
#define OBJ_LIST 1 /* List object. */
|
||||
#define OBJ_SET 2 /* Set object. */
|
||||
#define OBJ_ZSET 3 /* Sorted set object. */
|
||||
#define OBJ_HASH 4 /* Hash object. */
|
||||
|
||||
#define htonu64(v) intrev64(v)
|
||||
#define ntohu64(v) intrev64(v)
|
||||
|
||||
#define LLONG_MAX __LONG_LONG_MAX__
|
||||
|
||||
int rm_rdbLoadRioWithLoading(migrateObj *mobj);
|
||||
int rmRead(migrateObj *mobj, void *buf, size_t len);
|
||||
|
||||
int rm_rdbLoadType(migrateObj *mobj);
|
||||
int rmLoadRioWithLoading(migrateObj *mobj);
|
||||
|
||||
time_t rm_rdbLoadTime(migrateObj *mobj);
|
||||
int rmLoadType(migrateObj *mobj);
|
||||
|
||||
long long rm_rdbLoadMillisecondTime(migrateObj *mobj, int rdbver);
|
||||
time_t rmLoadTime(migrateObj *mobj);
|
||||
|
||||
int rm_rdbLoadLenByRef(migrateObj *mobi, int *isencoded, uint64_t *lenptr);
|
||||
long long rmLoadMillisecondTime(migrateObj *mobj, int rdbver);
|
||||
|
||||
int rmLoadLenByRef(migrateObj *mobi, int *isencoded, uint64_t *lenptr);
|
||||
|
||||
void rm_memrev64(void *p);
|
||||
|
||||
uint64_t rm_rdbLoadLen(migrateObj *mobj, int *isencoded);
|
||||
uint64_t rmLoadLen(migrateObj *mobj, int *isencoded);
|
||||
|
||||
void *rm_rdbGenericLoadStringObject(migrateObj *mobj, int flags, size_t *lenptr);
|
||||
sds rmLoadStringObject(migrateObj *mobj);
|
||||
|
||||
sds rmGenericLoadStringObject(migrateObj *mobj, int flags, size_t *lenptr);
|
||||
|
||||
void *rmLoadIntegerObject(migrateObj *mobj, int enctype, int flags, size_t *lenptr);
|
||||
|
||||
void *rmLoadLzfStringObject(migrateObj *mobj, int flags, size_t *lenptr);
|
||||
|
||||
robj *rmLoadObject(int rdbtype, migrateObj *mobj, sds key, int *error);
|
||||
|
||||
#endif
|
||||
|
||||
@ -147,9 +147,7 @@ void readFullData() {
|
||||
goto error;
|
||||
}
|
||||
if (buf[0] == '-') {
|
||||
serverLog(LL_WARNING,
|
||||
"MASTER aborted replication with an error: %s",
|
||||
buf + 1);
|
||||
serverLog(LL_WARNING, "MASTER aborted replication with an error: %s", buf + 1);
|
||||
goto error;
|
||||
} else if (buf[0] == '\0') {
|
||||
// mobj->repl_transfer_lastio = server.unixtime;
|
||||
@ -176,7 +174,7 @@ void readFullData() {
|
||||
(long long)mobj->repl_transfer_size);
|
||||
}
|
||||
}
|
||||
int flag = rdbLoadRioWithLoading(mobj);
|
||||
int flag = rmLoadRioWithLoading(mobj);
|
||||
|
||||
return;
|
||||
error:
|
||||
@ -292,21 +290,30 @@ void syncDataWithRedis(int fd, void *user_data, int mask) {
|
||||
mobj->repl_stat = REPL_STATE_RECEIVE_PSYNC_REPLY;
|
||||
return;
|
||||
}
|
||||
if (mobj->repl_stat != REPL_STATE_RECEIVE_PSYNC_REPLY) {
|
||||
serverLog(LL_WARNING, "syncDataWithRedis(): state machine error, state should be RECEIVE_PSYNC but is %d",
|
||||
mobj->repl_stat);
|
||||
goto error;
|
||||
}
|
||||
int psync_result = receiveDataFromRedis();
|
||||
if (psync_result == PSYNC_WAIT_REPLY)
|
||||
return;
|
||||
if (psync_result == PSYNC_TRY_LATER)
|
||||
goto error;
|
||||
if (psync_result == PSYNC_CONTINUE) {
|
||||
|
||||
if (mobj->repl_stat == REPL_STATE_RECEIVE_PSYNC_REPLY) {
|
||||
int psync_result = receiveDataFromRedis();
|
||||
if (psync_result == PSYNC_WAIT_REPLY)
|
||||
return;
|
||||
if (psync_result == PSYNC_TRY_LATER)
|
||||
goto error;
|
||||
if (psync_result == PSYNC_CONTINUE) {
|
||||
mobj->repl_stat = REPL_STATE_CONTINUE_SYNC;
|
||||
} else {
|
||||
mobj->repl_stat = REPL_STATE_FULL_SYNC;
|
||||
}
|
||||
}
|
||||
|
||||
// 接受全部数据
|
||||
serverLog(LL_NOTICE, "begin receive full data");
|
||||
readFullData();
|
||||
if (mobj->repl_stat == REPL_STATE_FULL_SYNC) {
|
||||
serverLog(LL_NOTICE, "begin receive full data");
|
||||
mobj->repl_stat = REPL_STATE_READING_FULL_DATA;
|
||||
readFullData();
|
||||
}
|
||||
if (mobj->repl_stat == REPL_STATE_CONTINUE_SYNC) {
|
||||
serverLog(LL_NOTICE, "begin receive continue data");
|
||||
}
|
||||
|
||||
return;
|
||||
error:
|
||||
if (err != NULL) {
|
||||
|
||||
@ -68,6 +68,9 @@ typedef enum
|
||||
REPL_STATE_RECEIVE_CAPA_REPLY, /* Wait for REPLCONF reply */
|
||||
REPL_STATE_SEND_PSYNC, /* Send PSYNC */
|
||||
REPL_STATE_RECEIVE_PSYNC_REPLY, /* Wait for PSYNC reply */
|
||||
REPL_STATE_FULL_SYNC,
|
||||
REPL_STATE_READING_FULL_DATA,
|
||||
REPL_STATE_CONTINUE_SYNC,
|
||||
/* --- End of handshake states --- */
|
||||
REPL_STATE_TRANSFER, /* Receiving .rdb from master */
|
||||
REPL_STATE_CONNECTED, /* Connected to master */
|
||||
|
||||
Loading…
Reference in New Issue
Block a user