这个文件我在今天分析学习的时候，一直有种似懂非懂的感觉，代码量700+的代码，最后开放给系统的就是一个process()方法。这里说的说的数据库检测，是针对key的检测，会用到，下面提到的结构体：
~~~
/* Data type to hold opcode with optional key name an success status */
/* 用于key的检测时使用，后续检测操作都用到了entry结构体 */
typedef struct {
//key的名字
char* key;
//类型
int type;
//是否是成功状态
char success;
} entry;
~~~
后续所涉及到的很多API都是与这个结构体相关，此代码最终检测的其实是一个叫dump.rdb的文件，在检测的后面还会加上循环冗余校验CRC64。下面亮出API：
~~~
int checkType(unsigned char t) /* 每当添加一个新的obj类型时，都要检测这个类型是否合理 */
int readBytes(void *target, long num) /* 在当前文件偏移量位置往后读取num个字节位置 */
int processHeader(void) /* 读取快照文件的头部，检测头部名称或版本号是否正确 */
int loadType(entry *e) /* 为entry赋上obj的Type */
int peekType() /* 弹出版本号 */
int processTime(int type) /* 去除用来表示时间的字节 */
uint32_t loadLength(int *isencoded) /* 分type读取长度 */
char *loadIntegerObject(int enctype) /* 根据当前整型的编码方式，获取数值，以字符形式返回 */
char* loadLzfStringObject() /* 获得解压后的字符串 */
char* loadStringObject() /* 获取当前文件信息字符串对象 */
int processStringObject(char**store) /* 将字符串对象赋给所传入的参数 */
double* loadDoubleValue() /* 文件中读取double类型值 */
int processDoubleValue(double**store) /* 对double类型进行赋予给参数 */
int loadPair(entry *e) /* 读取键值对 */
entry loadEntry() /* 获取entry的key结构体 */
void printCentered(int indent, int width, char* body) /* 输出界面对称的信息 */
void printValid(uint64_t ops, uint64_t bytes) /* 输出有效信息 */
void printSkipped(uint64_t bytes, uint64_t offset) /* 输出Skipped跳过bytes字节信息 */
void printErrorStack(entry *e) /* 输出错误栈的信息 */
void process(void) /* process方法是执行检测的主要方法 */
~~~
方法里面好多loadXXX()方法，这几个load方法的确比较有用，在这个检测文件中，编写者又很人性化的构造了error的结构体，用于模拟错误信息栈的输出。
~~~
/* Hold a stack of errors */
/* 错误信息结构体 */
typedef struct {
//具体的错误信息字符串
char error[16][1024];
//内部偏移量
size_t offset[16];
//错误信息等级
size_t level;
} errors_t;
static errors_t errors;
~~~
不同的level等级对应不同的出错信息。在API里有个比较关键的方法，loadEntry,获取key相关的结构体;
~~~
/* 获取entry的key结构体 */
entry loadEntry() {
entry e = { NULL, -1, 0 };
uint32_t length, offset[4];
/* reset error container */
errors.level = 0;
offset[0] = CURR_OFFSET;
//此处赋值type
if (!loadType(&e)) {
return e;
}
offset[1] = CURR_OFFSET;
if (e.type == REDIS_SELECTDB) {
if ((length = loadLength(NULL)) == REDIS_RDB_LENERR) {
SHIFT_ERROR(offset[1], "Error reading database number");
return e;
}
if (length > 63) {
SHIFT_ERROR(offset[1], "Database number out of range (%d)", length);
return e;
}
} else if (e.type == REDIS_EOF) {
if (positions[level].offset < positions[level].size) {
SHIFT_ERROR(offset[0], "Unexpected EOF");
} else {
e.success = 1;
}
return e;
} else {
/* optionally consume expire */
if (e.type == REDIS_EXPIRETIME ||
e.type == REDIS_EXPIRETIME_MS) {
if (!processTime(e.type)) return e;
if (!loadType(&e)) return e;
}
offset[1] = CURR_OFFSET;
//调用loadPair为Entry赋值key
if (!loadPair(&e)) {
SHIFT_ERROR(offset[1], "Error for type %s", types[e.type]);
return e;
}
}
/* all entries are followed by a valid type:
* e.g. a new entry, SELECTDB, EXPIRE, EOF */
offset[2] = CURR_OFFSET;
if (peekType() == -1) {
SHIFT_ERROR(offset[2], "Followed by invalid type");
SHIFT_ERROR(offset[0], "Error for type %s", types[e.type]);
e.success = 0;
} else {
e.success = 1;
}
return e;
}
~~~
其中里面的关键的赋值key,value在loadPair()方法:
~~~
/* 读取键值对 */
int loadPair(entry *e) {
uint32_t offset = CURR_OFFSET;
uint32_t i;
/* read key first */
//首先从文件中读取key值
char *key;
if (processStringObject(&key)) {
e->key = key;
} else {
SHIFT_ERROR(offset, "Error reading entry key");
return 0;
}
uint32_t length = 0;
if (e->type == REDIS_LIST ||
e->type == REDIS_SET ||
e->type == REDIS_ZSET ||
e->type == REDIS_HASH) {
if ((length = loadLength(NULL)) == REDIS_RDB_LENERR) {
SHIFT_ERROR(offset, "Error reading %s length", types[e->type]);
return 0;
}
}
//读取key值后面跟着的value值
switch(e->type) {
case REDIS_STRING:
case REDIS_HASH_ZIPMAP:
case REDIS_LIST_ZIPLIST:
case REDIS_SET_INTSET:
case REDIS_ZSET_ZIPLIST:
case REDIS_HASH_ZIPLIST:
//因为类似ziplist,zipmap等结构体其实是一个个结点连接而成的超级字符串，所以是直接读取
if (!processStringObject(NULL)) {
SHIFT_ERROR(offset, "Error reading entry value");
return 0;
}
break;
case REDIS_LIST:
case REDIS_SET:
//而上面这2种是传统的结构，要分结点读取
for (i = 0; i < length; i++) {
offset = CURR_OFFSET;
if (!processStringObject(NULL)) {
SHIFT_ERROR(offset, "Error reading element at index %d (length: %d)", i, length);
return 0;
}
}
break;
case REDIS_ZSET:
for (i = 0; i < length; i++) {
offset = CURR_OFFSET;
if (!processStringObject(NULL)) {
SHIFT_ERROR(offset, "Error reading element key at index %d (length: %d)", i, length);
return 0;
}
offset = CURR_OFFSET;
if (!processDoubleValue(NULL)) {
SHIFT_ERROR(offset, "Error reading element value at index %d (length: %d)", i, length);
return 0;
}
}
break;
case REDIS_HASH:
for (i = 0; i < length; i++) {
offset = CURR_OFFSET;
if (!processStringObject(NULL)) {
SHIFT_ERROR(offset, "Error reading element key at index %d (length: %d)", i, length);
return 0;
}
offset = CURR_OFFSET;
if (!processStringObject(NULL)) {
SHIFT_ERROR(offset, "Error reading element value at index %d (length: %d)", i, length);
return 0;
}
}
break;
default:
SHIFT_ERROR(offset, "Type not implemented");
return 0;
}
/* because we're done, we assume success */
//只要执行过了，我们就认定为成功
e->success = 1;
return 1;
}
~~~
如果e-success=1则说明这个key的检测就过关了。为什么这么说呢，我们来看主检测方法process()方法：
~~~
/* process方法是执行检测的主要方法 */
void process(void) {
uint64_t num_errors = 0, num_valid_ops = 0, num_valid_bytes = 0;
entry entry;
//读取文件头部获取快照文件版本号
int dump_version = processHeader();
/* Exclude the final checksum for RDB >= 5. Will be checked at the end. */
if (dump_version >= 5) {
if (positions[0].size < 8) {
printf("RDB version >= 5 but no room for checksum.\n");
exit(1);
}
positions[0].size -= 8;
}
level = 1;
while(positions[0].offset < positions[0].size) {
positions[1] = positions[0];
entry = loadEntry();
if (!entry.success) {
//如果Entry不为成功状态
printValid(num_valid_ops, num_valid_bytes);
printErrorStack(&entry);
num_errors++;
num_valid_ops = 0;
num_valid_bytes = 0;
/* search for next valid entry */
uint64_t offset = positions[0].offset + 1;
int i = 0;
//接着寻找后面3个有效entries
while (!entry.success && offset < positions[0].size) {
positions[1].offset = offset;
/* find 3 consecutive valid entries */
//寻找3个有效的entries
for (i = 0; i < 3; i++) {
entry = loadEntry();
if (!entry.success) break;
}
/* check if we found 3 consecutive valid entries */
if (i < 3) {
offset++;
}
}
/* print how many bytes we have skipped to find a new valid opcode */
if (offset < positions[0].size) {
printSkipped(offset - positions[0].offset, offset);
}
positions[0].offset = offset;
} else {
num_valid_ops++;
num_valid_bytes += positions[1].offset - positions[0].offset;
/* advance position */
positions[0] = positions[1];
}
free(entry.key);
}
/* because there is another potential error,
* print how many valid ops we have processed */
printValid(num_valid_ops, num_valid_bytes);
/* expect an eof */
if (entry.type != REDIS_EOF) {
/* last byte should be EOF, add error */
errors.level = 0;
SHIFT_ERROR(positions[0].offset, "Expected EOF, got %s", types[entry.type]);
/* this is an EOF error so reset type */
entry.type = -1;
printErrorStack(&entry);
num_errors++;
}
/* Verify checksum */
//版本号>=5的时候，需要检验校验和
if (dump_version >= 5) {
uint64_t crc = crc64(0,positions[0].data,positions[0].size);
uint64_t crc2;
unsigned char *p = (unsigned char*)positions[0].data+positions[0].size;
crc2 = ((uint64_t)p[0] << 0) |
((uint64_t)p[1] << 8) |
((uint64_t)p[2] << 16) |
((uint64_t)p[3] << 24) |
((uint64_t)p[4] << 32) |
((uint64_t)p[5] << 40) |
((uint64_t)p[6] << 48) |
((uint64_t)p[7] << 56);
if (crc != crc2) {
SHIFT_ERROR(positions[0].offset, "RDB CRC64 does not match.");
} else {
printf("CRC64 checksum is OK\n");
}
}
/* print summary on errors */
if (num_errors) {
printf("\n");
printf("Total unprocessable opcodes: %llu\n",
(unsigned long long) num_errors);
}
}
~~~
如果想了解检测的详细原理，事先了解dump.rdb的文件内容结构也许会对我们又很大帮助。
- 前言
- （一）--Redis结构解析
- （二）--结构体分析(1)
- （三）---dict哈希结构
- （四）-- sds字符串
- （五）--- sparkline微线图
- （六）--- ziplist压缩列表
- （七）--- zipmap压缩图
- （八）--- t_hash哈希转换
- （九）--- t_list，t_string的分析
- （十）--- testhelp.h小型测试框架和redis-check-aof.c日志检测
- （十一）--- memtest内存检测
- （十二）--- redis-check-dump本地数据库检测
- （十三）--- redis-benchmark性能测试
- （十四）--- rdb.c本地数据库操作
- （十五）--- aof-append only file解析
- （十六）--- config配置文件
- （十七）--- multi事务操作
- （十八）--- db.c内存数据库操作
- （十九）--- replication主从数据复制的实现
- （二十）--- ae事件驱动
- （二十一）--- anet网络通信的封装
- （二十二）--- networking网络协议传输
- （二十三）--- CRC循环冗余算法和RAND随机数算法
- （二十四）--- tool工具类（2）
- （二十五）--- zmalloc内存分配实现
- （二十六）--- slowLog和hyperloglog
- （二十七）--- rio系统I/O的封装
- （二十八）--- object创建和释放redisObject对象
- （二十九）--- bio后台I/O服务的实现
- （三十）--- pubsub发布订阅模式
- （三十一）--- latency延迟分析处理
- （三十二）--- redis-cli.c客户端命令行接口的实现(1)
- （三十三）--- redis-cli.c客户端命令行接口的实现(2)
- （三十四）--- redis.h服务端的实现分析（1）
- （三十五）--- redis.c服务端的实现分析（2）
- （三十六）--- Redis中的11大优秀设计