xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

make wal meta serialized and flushed into disk

This commit is contained in:
Liu Jicong 2021-12-10 11:31:36 +08:00
parent 713c20c43d
commit bd408504bb
4 changed files with 58 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/libs/wal/inc/walInt.h
View File

@ -83,8 +83,8 @@ int walReadMeta(SWal* pWal);
int walWriteMeta(SWal* pWal); int walWriteMeta(SWal* pWal);
int walRollFileInfo(SWal* pWal); int walRollFileInfo(SWal* pWal);
char* walFileInfoSerialize(SWal* pWal); char* walMetaSerialize(SWal* pWal);
SArray* walFileInfoDeserialize(const char* bytes); int walMetaDeserialize(SWal* pWal, const char* bytes);
//meta section end //meta section end
int64_t walGetSeq(); int64_t walGetSeq();

80
source/libs/wal/src/walMeta.c
View File

@ -47,49 +47,74 @@ int walRollFileInfo(SWal* pWal) {
return 0; return 0;
} }
char* walFileInfoSerialize(SWal* pWal) { char* walMetaSerialize(SWal* pWal) {
char buf[30]; char buf[30];
if(pWal == NULL || pWal->fileInfoSet == NULL) return 0; if(pWal == NULL || pWal->fileInfoSet == NULL) return 0;
int sz = pWal->fileInfoSet->size; int sz = pWal->fileInfoSet->size;
cJSON* root = cJSON_CreateArray(); cJSON* pRoot = cJSON_CreateObject();
cJSON* field; cJSON* pMeta = cJSON_CreateObject();
if(root == NULL) { cJSON* pFiles = cJSON_CreateArray();
cJSON* pField;
if(pRoot == NULL || pMeta == NULL || pFiles == NULL) {
//TODO //TODO
return NULL; return NULL;
} }
cJSON_AddItemToObject(pRoot, "meta", pMeta);
sprintf(buf, "%" PRId64, pWal->firstVersion);
cJSON_AddStringToObject(pMeta, "firstVer", buf);
sprintf(buf, "%" PRId64, pWal->snapshotVersion);
cJSON_AddStringToObject(pMeta, "snapshotVer", buf);
sprintf(buf, "%" PRId64, pWal->commitVersion);
cJSON_AddStringToObject(pMeta, "commitVer", buf);
sprintf(buf, "%" PRId64, pWal->lastVersion);
cJSON_AddStringToObject(pMeta, "lastVer", buf);
cJSON_AddItemToObject(pRoot, "files", pFiles);
WalFileInfo* pData = pWal->fileInfoSet->pData; WalFileInfo* pData = pWal->fileInfoSet->pData;
for(int i = 0; i < sz; i++) { for(int i = 0; i < sz; i++) {
WalFileInfo* pInfo = &pData[i]; WalFileInfo* pInfo = &pData[i];
cJSON_AddItemToArray(root, field = cJSON_CreateObject()); cJSON_AddItemToArray(pFiles, pField = cJSON_CreateObject());
if(field == NULL) { if(pField == NULL) {
cJSON_Delete(root); cJSON_Delete(pRoot);
return NULL; return NULL;
} }
//cjson only support int32_t or double //cjson only support int32_t or double
//string are used to prohibit the loss of precision //string are used to prohibit the loss of precision
sprintf(buf, "%ld", pInfo->firstVer); sprintf(buf, "%" PRId64, pInfo->firstVer);
cJSON_AddStringToObject(field, "firstVer", buf); cJSON_AddStringToObject(pField, "firstVer", buf);
sprintf(buf, "%ld", pInfo->lastVer); sprintf(buf, "%" PRId64, pInfo->lastVer);
cJSON_AddStringToObject(field, "lastVer", buf); cJSON_AddStringToObject(pField, "lastVer", buf);
sprintf(buf, "%ld", pInfo->createTs); sprintf(buf, "%" PRId64, pInfo->createTs);
cJSON_AddStringToObject(field, "createTs", buf); cJSON_AddStringToObject(pField, "createTs", buf);
sprintf(buf, "%ld", pInfo->closeTs); sprintf(buf, "%" PRId64, pInfo->closeTs);
cJSON_AddStringToObject(field, "closeTs", buf); cJSON_AddStringToObject(pField, "closeTs", buf);
sprintf(buf, "%ld", pInfo->fileSize); sprintf(buf, "%" PRId64, pInfo->fileSize);
cJSON_AddStringToObject(field, "fileSize", buf); cJSON_AddStringToObject(pField, "fileSize", buf);
} }
return cJSON_Print(root); return cJSON_Print(pRoot);
} }
SArray* walFileInfoDeserialize(const char* bytes) { int walMetaDeserialize(SWal* pWal, const char* bytes) {
cJSON *root, *pInfoJson, *pField; ASSERT(taosArrayGetSize(pWal->fileInfoSet) == 0);
root = cJSON_Parse(bytes); cJSON *pRoot, *pMeta, *pFiles, *pInfoJson, *pField;
int sz = cJSON_GetArraySize(root); pRoot = cJSON_Parse(bytes);
pMeta = cJSON_GetObjectItem(pRoot, "meta");
pField = cJSON_GetObjectItem(pMeta, "firstVer");
pWal->firstVersion = atoll(cJSON_GetStringValue(pField));
pField = cJSON_GetObjectItem(pMeta, "snapshotVer");
pWal->snapshotVersion = atoll(cJSON_GetStringValue(pField));
pField = cJSON_GetObjectItem(pMeta, "commitVer");
pWal->commitVersion = atoll(cJSON_GetStringValue(pField));
pField = cJSON_GetObjectItem(pMeta, "lastVer");
pWal->lastVersion = atoll(cJSON_GetStringValue(pField));
pFiles = cJSON_GetObjectItem(pRoot, "files");
int sz = cJSON_GetArraySize(pFiles);
//deserialize //deserialize
SArray* pArray = taosArrayInit(sz, sizeof(WalFileInfo)); SArray* pArray = taosArrayInit(sz, sizeof(WalFileInfo));
WalFileInfo *pData = pArray->pData; WalFileInfo *pData = pArray->pData;
for(int i = 0; i < sz; i++) { for(int i = 0; i < sz; i++) {
cJSON* pInfoJson = cJSON_GetArrayItem(root, i); cJSON* pInfoJson = cJSON_GetArrayItem(pFiles, i);
WalFileInfo* pInfo = &pData[i]; WalFileInfo* pInfo = &pData[i];
pField = cJSON_GetObjectItem(pInfoJson, "firstVer"); pField = cJSON_GetObjectItem(pInfoJson, "firstVer");
pInfo->firstVer = atoll(cJSON_GetStringValue(pField)); pInfo->firstVer = atoll(cJSON_GetStringValue(pField));
@ -103,7 +128,8 @@ SArray* walFileInfoDeserialize(const char* bytes) {
pInfo->fileSize = atoll(cJSON_GetStringValue(pField)); pInfo->fileSize = atoll(cJSON_GetStringValue(pField));
} }
taosArraySetSize(pArray, sz); taosArraySetSize(pArray, sz);
return pArray; pWal->fileInfoSet = pArray;
return 0;
} }
static inline int walBuildMetaName(SWal* pWal, int metaVer, char* buf) { static inline int walBuildMetaName(SWal* pWal, int metaVer, char* buf) {
@ -144,7 +170,7 @@ int walWriteMeta(SWal* pWal) {
if(metaTfd < 0) { if(metaTfd < 0) {
return -1; return -1;
} }
char* serialized = walFileInfoSerialize(pWal); char* serialized = walMetaSerialize(pWal);
int len = strlen(serialized); int len = strlen(serialized);
if(len != tfWrite(metaTfd, serialized, len)) { if(len != tfWrite(metaTfd, serialized, len)) {
//TODO:clean file //TODO:clean file
@ -183,8 +209,8 @@ int walReadMeta(SWal* pWal) {
return -1; return -1;
} }
//load into fileInfoSet //load into fileInfoSet
pWal->fileInfoSet = walFileInfoDeserialize(buf); int code = walMetaDeserialize(pWal, buf);
if(pWal->fileInfoSet == NULL) { if(code != 0) {
free(buf); free(buf);
return -1; return -1;
} }

2
source/libs/wal/src/walMgmt.c
View File

@ -175,8 +175,6 @@ static void walFreeObj(void *wal) {
tfClose(pWal->writeIdxTfd); tfClose(pWal->writeIdxTfd);
taosArrayDestroy(pWal->fileInfoSet); taosArrayDestroy(pWal->fileInfoSet);
pWal->fileInfoSet = NULL; pWal->fileInfoSet = NULL;
taosArrayDestroy(pWal->fileInfoSet);
pWal->fileInfoSet = NULL;
pthread_mutex_destroy(&pWal->mutex); pthread_mutex_destroy(&pWal->mutex);
tfree(pWal); tfree(pWal);
} }

7
source/libs/wal/test/walMetaTest.cpp
View File

@ -92,7 +92,7 @@ TEST_F(WalCleanEnv, serialize) {
ASSERT(code == 0); ASSERT(code == 0);
code = walRollFileInfo(pWal); code = walRollFileInfo(pWal);
ASSERT(code == 0); ASSERT(code == 0);
char*ss = walFileInfoSerialize(pWal); char*ss = walMetaSerialize(pWal);
printf("%s\n", ss); printf("%s\n", ss);
code = walWriteMeta(pWal); code = walWriteMeta(pWal);
ASSERT(code == 0); ASSERT(code == 0);
@ -113,20 +113,19 @@ TEST_F(WalCleanEnv, removeOldMeta) {
TEST_F(WalKeepEnv, readOldMeta) { TEST_F(WalKeepEnv, readOldMeta) {
int code = walRollFileInfo(pWal); int code = walRollFileInfo(pWal);
ASSERT(code == 0); ASSERT(code == 0);
ASSERT(pWal->fileInfoSet != NULL);
code = walWriteMeta(pWal); code = walWriteMeta(pWal);
ASSERT(code == 0); ASSERT(code == 0);
code = walRollFileInfo(pWal); code = walRollFileInfo(pWal);
ASSERT(code == 0); ASSERT(code == 0);
code = walWriteMeta(pWal); code = walWriteMeta(pWal);
ASSERT(code == 0); ASSERT(code == 0);
char*oldss = walFileInfoSerialize(pWal); char*oldss = walMetaSerialize(pWal);
TearDown(); TearDown();
SetUp(); SetUp();
code = walReadMeta(pWal); code = walReadMeta(pWal);
ASSERT(code == 0); ASSERT(code == 0);
char* newss = walFileInfoSerialize(pWal); char* newss = walMetaSerialize(pWal);
int len = strlen(oldss); int len = strlen(oldss);
ASSERT_EQ(len, strlen(newss)); ASSERT_EQ(len, strlen(newss));