? ? ?這個文件我在今天分析學習的時候，一直有種似懂非懂的感覺，代碼量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大優秀設計