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

Merge pull request #26788 from taosdata/fix/TD-30990 

void unused return values
This commit is contained in:
Hongze Cheng 2024-07-27 17:08:28 +08:00 committed by GitHub
parent 805a1387b7 65222d9f2a
commit 89e4972acc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 325 additions and 325 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
source/common/src/cos.c
View File

@ -65,13 +65,13 @@ int32_t s3CheckCfg() {
int32_t code = 0, lino = 0; int32_t code = 0, lino = 0;
if (!tsS3Enabled) { if (!tsS3Enabled) {
fprintf(stderr, "s3 not configured.\n"); (void)fprintf(stderr, "s3 not configured.\n");
goto _exit; goto _exit;
} }
code = s3Begin(); code = s3Begin();
if (code != 0) { if (code != 0) {
fprintf(stderr, "failed to initialize s3.\n"); (void)fprintf(stderr, "failed to initialize s3.\n");
TAOS_CHECK_GOTO(code, &lino, _exit); TAOS_CHECK_GOTO(code, &lino, _exit);
} }
@ -82,72 +82,72 @@ int32_t s3CheckCfg() {
int ds_len = strlen(TD_DIRSEP); int ds_len = strlen(TD_DIRSEP);
int tmp_len = strlen(tsTempDir); int tmp_len = strlen(tsTempDir);
snprintf(path, PATH_MAX, "%s", tsTempDir); (void)snprintf(path, PATH_MAX, "%s", tsTempDir);
if (strncmp(tsTempDir + tmp_len - ds_len, TD_DIRSEP, ds_len) != 0) { if (strncmp(tsTempDir + tmp_len - ds_len, TD_DIRSEP, ds_len) != 0) {
snprintf(path + tmp_len, PATH_MAX - tmp_len, "%s", TD_DIRSEP); (void)snprintf(path + tmp_len, PATH_MAX - tmp_len, "%s", TD_DIRSEP);
snprintf(path + tmp_len + ds_len, PATH_MAX - tmp_len - ds_len, "%s", objectname[0]); (void)snprintf(path + tmp_len + ds_len, PATH_MAX - tmp_len - ds_len, "%s", objectname[0]);
} else { } else {
snprintf(path + tmp_len, PATH_MAX - tmp_len, "%s", objectname[0]); (void)snprintf(path + tmp_len, PATH_MAX - tmp_len, "%s", objectname[0]);
} }
TdFilePtr fp = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC); TdFilePtr fp = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
if (!fp) { if (!fp) {
fprintf(stderr, "failed to open test file: %s.\n", path); (void)fprintf(stderr, "failed to open test file: %s.\n", path);
// uError("ERROR: %s Failed to open %s", __func__, path); // uError("ERROR: %s Failed to open %s", __func__, path);
TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit); TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit);
} }
if (taosWriteFile(fp, testdata, strlen(testdata)) < 0) { if (taosWriteFile(fp, testdata, strlen(testdata)) < 0) {
fprintf(stderr, "failed to write test file: %s.\n", path); (void)fprintf(stderr, "failed to write test file: %s.\n", path);
TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit); TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit);
} }
if (taosFsyncFile(fp) < 0) { if (taosFsyncFile(fp) < 0) {
fprintf(stderr, "failed to fsync test file: %s.\n", path); (void)fprintf(stderr, "failed to fsync test file: %s.\n", path);
TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit); TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit);
} }
taosCloseFile(&fp); (void)taosCloseFile(&fp);
fprintf(stderr, "\nstart to put object: %s, file: %s content: %s\n", objectname[0], path, testdata); (void)fprintf(stderr, "\nstart to put object: %s, file: %s content: %s\n", objectname[0], path, testdata);
code = s3PutObjectFromFileOffset(path, objectname[0], 0, 16); code = s3PutObjectFromFileOffset(path, objectname[0], 0, 16);
if (code != 0) { if (code != 0) {
fprintf(stderr, "put object %s : failed.\n", objectname[0]); (void)fprintf(stderr, "put object %s : failed.\n", objectname[0]);
TAOS_CHECK_GOTO(code, &lino, _exit); TAOS_CHECK_GOTO(code, &lino, _exit);
} }
fprintf(stderr, "put object %s: success.\n\n", objectname[0]); (void)fprintf(stderr, "put object %s: success.\n\n", objectname[0]);
// list buckets // list buckets
fprintf(stderr, "start to list bucket %s by prefix s3.\n", tsS3BucketName); (void)fprintf(stderr, "start to list bucket %s by prefix s3.\n", tsS3BucketName);
code = s3ListBucket(tsS3BucketName); code = s3ListBucket(tsS3BucketName);
if (code != 0) { if (code != 0) {
fprintf(stderr, "listing bucket %s : failed.\n", tsS3BucketName); (void)fprintf(stderr, "listing bucket %s : failed.\n", tsS3BucketName);
TAOS_CHECK_GOTO(code, &lino, _exit); TAOS_CHECK_GOTO(code, &lino, _exit);
} }
fprintf(stderr, "listing bucket %s: success.\n\n", tsS3BucketName); (void)fprintf(stderr, "listing bucket %s: success.\n\n", tsS3BucketName);
// test range get // test range get
uint8_t *pBlock = NULL; uint8_t *pBlock = NULL;
int c_offset = 10; int c_offset = 10;
int c_len = 6; int c_len = 6;
fprintf(stderr, "start to range get object %s offset: %d len: %d.\n", objectname[0], c_offset, c_len); (void)fprintf(stderr, "start to range get object %s offset: %d len: %d.\n", objectname[0], c_offset, c_len);
code = s3GetObjectBlock(objectname[0], c_offset, c_len, true, &pBlock); code = s3GetObjectBlock(objectname[0], c_offset, c_len, true, &pBlock);
if (code != 0) { if (code != 0) {
fprintf(stderr, "get object %s : failed.\n", objectname[0]); (void)fprintf(stderr, "get object %s : failed.\n", objectname[0]);
TAOS_CHECK_GOTO(code, &lino, _exit); TAOS_CHECK_GOTO(code, &lino, _exit);
} }
char buf[7] = {0}; char buf[7] = {0};
memcpy(buf, pBlock, c_len); (void)memcpy(buf, pBlock, c_len);
taosMemoryFree(pBlock); taosMemoryFree(pBlock);
fprintf(stderr, "object content: %s\n", buf); (void)fprintf(stderr, "object content: %s\n", buf);
fprintf(stderr, "get object %s: success.\n\n", objectname[0]); (void)fprintf(stderr, "get object %s: success.\n\n", objectname[0]);
// delete test object // delete test object
fprintf(stderr, "start to delete object: %s.\n", objectname[0]); (void)fprintf(stderr, "start to delete object: %s.\n", objectname[0]);
code = s3DeleteObjects(objectname, 1); code = s3DeleteObjects(objectname, 1);
if (code != 0) { if (code != 0) {
fprintf(stderr, "delete object %s : failed.\n", objectname[0]); (void)fprintf(stderr, "delete object %s : failed.\n", objectname[0]);
TAOS_CHECK_GOTO(code, &lino, _exit); TAOS_CHECK_GOTO(code, &lino, _exit);
} }
fprintf(stderr, "delete object %s: success.\n\n", objectname[0]); (void)fprintf(stderr, "delete object %s: success.\n\n", objectname[0]);
s3End(); s3End();
@ -252,9 +252,9 @@ static int32_t s3ListBucket(char const *bucketname) {
const char **object_name = TARRAY_DATA(objectArray); const char **object_name = TARRAY_DATA(objectArray);
int size = TARRAY_SIZE(objectArray); int size = TARRAY_SIZE(objectArray);
fprintf(stderr, "objects:\n"); (void)fprintf(stderr, "objects:\n");
for (int i = 0; i < size; ++i) { for (int i = 0; i < size; ++i) {
fprintf(stderr, "%s\n", object_name[i]); (void)fprintf(stderr, "%s\n", object_name[i]);
} }
taosArrayDestroyEx(objectArray, s3FreeObjectKey); taosArrayDestroyEx(objectArray, s3FreeObjectKey);
@ -300,7 +300,7 @@ static int growbuffer_append(growbuffer **gb, const char *data, int dataLen) {
toCopy = dataLen; toCopy = dataLen;
} }
memcpy(&(buf->data[buf->size]), data, toCopy); (void)memcpy(&(buf->data[buf->size]), data, toCopy);
buf->size += toCopy, data += toCopy, dataLen -= toCopy; buf->size += toCopy, data += toCopy, dataLen -= toCopy;
} }
@ -319,7 +319,7 @@ static void growbuffer_read(growbuffer **gb, int amt, int *amtReturn, char *buff
*amtReturn = (buf->size > amt) ? amt : buf->size; *amtReturn = (buf->size > amt) ? amt : buf->size;
memcpy(buffer, &(buf->data[buf->start]), *amtReturn); (void)memcpy(buffer, &(buf->data[buf->start]), *amtReturn);
buf->start += *amtReturn, buf->size -= *amtReturn; buf->start += *amtReturn, buf->size -= *amtReturn;
@ -440,7 +440,7 @@ S3Status initial_multipart_callback(const char *upload_id, void *callbackData) {
} }
S3Status MultipartResponseProperiesCallback(const S3ResponseProperties *properties, void *callbackData) { S3Status MultipartResponseProperiesCallback(const S3ResponseProperties *properties, void *callbackData) {
responsePropertiesCallbackNull(properties, callbackData); (void)responsePropertiesCallbackNull(properties, callbackData);
MultipartPartData *data = (MultipartPartData *)callbackData; MultipartPartData *data = (MultipartPartData *)callbackData;
int seq = data->seq; int seq = data->seq;
@ -451,7 +451,7 @@ S3Status MultipartResponseProperiesCallback(const S3ResponseProperties *properti
} }
S3Status MultipartResponseProperiesCallbackWithCp(const S3ResponseProperties *properties, void *callbackData) { S3Status MultipartResponseProperiesCallbackWithCp(const S3ResponseProperties *properties, void *callbackData) {
responsePropertiesCallbackNull(properties, callbackData); (void)responsePropertiesCallbackNull(properties, callbackData);
MultipartPartData *data = (MultipartPartData *)callbackData; MultipartPartData *data = (MultipartPartData *)callbackData;
int seq = data->seq; int seq = data->seq;
@ -624,7 +624,7 @@ static int32_t s3PutObjectFromFileWithoutCp(S3BucketContext *bucket_context, cha
} }
MultipartPartData partData; MultipartPartData partData;
memset(&partData, 0, sizeof(MultipartPartData)); (void)memset(&partData, 0, sizeof(MultipartPartData));
int partContentLength = 0; int partContentLength = 0;
S3MultipartInitialHandler handler = {{&responsePropertiesCallbackNull, &responseCompleteCallback}, S3MultipartInitialHandler handler = {{&responsePropertiesCallbackNull, &responseCompleteCallback},
@ -739,7 +739,7 @@ static int32_t s3PutObjectFromFileWithCp(S3BucketContext *bucket_context, const
bool need_init_upload = true; bool need_init_upload = true;
char file_cp_path[TSDB_FILENAME_LEN]; char file_cp_path[TSDB_FILENAME_LEN];
snprintf(file_cp_path, TSDB_FILENAME_LEN, "%s.cp", file); (void)snprintf(file_cp_path, TSDB_FILENAME_LEN, "%s.cp", file);
SCheckpoint cp = {0}; SCheckpoint cp = {0};
cp.parts = taosMemoryCalloc(max_part_num, sizeof(SCheckpointPart)); cp.parts = taosMemoryCalloc(max_part_num, sizeof(SCheckpointPart));
@ -781,7 +781,7 @@ static int32_t s3PutObjectFromFileWithCp(S3BucketContext *bucket_context, const
// cos_cp_get_undo_parts(&cp, &part_num, parts, &consume_bytes); // cos_cp_get_undo_parts(&cp, &part_num, parts, &consume_bytes);
MultipartPartData partData; MultipartPartData partData;
memset(&partData, 0, sizeof(MultipartPartData)); (void)memset(&partData, 0, sizeof(MultipartPartData));
int partContentLength = 0; int partContentLength = 0;
S3PutObjectHandler putObjectHandler = {{&MultipartResponseProperiesCallbackWithCp, &responseCompleteCallback}, S3PutObjectHandler putObjectHandler = {{&MultipartResponseProperiesCallbackWithCp, &responseCompleteCallback},
@ -920,7 +920,7 @@ int32_t s3PutObjectFromFile2(const char *file, const char *object_name, int8_t w
char useServerSideEncryption = 0; char useServerSideEncryption = 0;
put_object_callback_data data = {0}; put_object_callback_data data = {0};
if (taosStatFile(file, &contentLength, &lmtime, NULL) < 0) { if (taosStatFile(file, (int64_t *)&contentLength, &lmtime, NULL) < 0) {
uError("ERROR: %s Failed to stat file %s: ", __func__, file); uError("ERROR: %s Failed to stat file %s: ", __func__, file);
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
@ -953,7 +953,7 @@ int32_t s3PutObjectFromFile2(const char *file, const char *object_name, int8_t w
} }
if (data.infileFD) { if (data.infileFD) {
taosCloseFile(&data.infileFD); (void)taosCloseFile(&data.infileFD);
} else if (data.gb) { } else if (data.gb) {
growbuffer_destroy(data.gb); growbuffer_destroy(data.gb);
} }
@ -975,7 +975,7 @@ int32_t s3PutObjectFromFileOffset(const char *file, const char *object_name, int
char useServerSideEncryption = 0; char useServerSideEncryption = 0;
put_object_callback_data data = {0}; put_object_callback_data data = {0};
if (taosStatFile(file, &contentLength, &lmtime, NULL) < 0) { if (taosStatFile(file, (int64_t *)&contentLength, &lmtime, NULL) < 0) {
uError("ERROR: %s Failed to stat file %s: ", __func__, file); uError("ERROR: %s Failed to stat file %s: ", __func__, file);
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
@ -987,7 +987,7 @@ int32_t s3PutObjectFromFileOffset(const char *file, const char *object_name, int
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
if (taosLSeekFile(data.infileFD, offset, SEEK_SET) < 0) { if (taosLSeekFile(data.infileFD, offset, SEEK_SET) < 0) {
taosCloseFile(&data.infileFD); (void)taosCloseFile(&data.infileFD);
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
@ -1010,7 +1010,7 @@ int32_t s3PutObjectFromFileOffset(const char *file, const char *object_name, int
} }
if (data.infileFD) { if (data.infileFD) {
taosCloseFile(&data.infileFD); (void)taosCloseFile(&data.infileFD);
} else if (data.gb) { } else if (data.gb) {
growbuffer_destroy(data.gb); growbuffer_destroy(data.gb);
} }
@ -1038,7 +1038,7 @@ static S3Status listBucketCallback(int isTruncated, const char *nextMarker, int
nextMarker = contents[contentsCount - 1].key; nextMarker = contents[contentsCount - 1].key;
} }
if (nextMarker) { if (nextMarker) {
snprintf(data->nextMarker, sizeof(data->nextMarker), "%s", nextMarker); (void)snprintf(data->nextMarker, sizeof(data->nextMarker), "%s", nextMarker);
} else { } else {
data->nextMarker[0] = 0; data->nextMarker[0] = 0;
} }
@ -1052,7 +1052,7 @@ static S3Status listBucketCallback(int isTruncated, const char *nextMarker, int
const S3ListBucketContent *content = &(contents[i]); const S3ListBucketContent *content = &(contents[i]);
// printf("%-50s", content->key); // printf("%-50s", content->key);
char *object_key = strdup(content->key); char *object_key = strdup(content->key);
taosArrayPush(data->objectArray, &object_key); (void)taosArrayPush(data->objectArray, &object_key);
} }
data->keyCount += contentsCount; data->keyCount += contentsCount;
@ -1139,7 +1139,7 @@ int32_t s3DeleteObjects(const char *object_name[], int nobject) {
void s3DeleteObjectsByPrefix(const char *prefix) { void s3DeleteObjectsByPrefix(const char *prefix) {
SArray *objectArray = getListByPrefix(prefix); SArray *objectArray = getListByPrefix(prefix);
if (objectArray == NULL) return; if (objectArray == NULL) return;
s3DeleteObjects(TARRAY_DATA(objectArray), TARRAY_SIZE(objectArray)); (void)s3DeleteObjects(TARRAY_DATA(objectArray), TARRAY_SIZE(objectArray));
taosArrayDestroyEx(objectArray, s3FreeObjectKey); taosArrayDestroyEx(objectArray, s3FreeObjectKey);
} }
@ -1156,7 +1156,7 @@ static S3Status getObjectDataCallback(int bufferSize, const char *buffer, void *
} }
if (cbd->buf) { if (cbd->buf) {
memcpy(cbd->buf + cbd->buf_pos, buffer, bufferSize); (void)memcpy(cbd->buf + cbd->buf_pos, buffer, bufferSize);
cbd->buf_pos += bufferSize; cbd->buf_pos += bufferSize;
cbd->status = S3StatusOK; cbd->status = S3StatusOK;
return S3StatusOK; return S3StatusOK;
@ -1196,7 +1196,7 @@ int32_t s3GetObjectBlock(const char *object_name, int64_t offset, int64_t size,
TAOS_RETURN(TAOS_SYSTEM_ERROR(EIO)); TAOS_RETURN(TAOS_SYSTEM_ERROR(EIO));
} }
*ppBlock = cbd.buf; *ppBlock = (uint8_t *)cbd.buf;
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
} }
@ -1231,11 +1231,11 @@ int32_t s3GetObjectToFile(const char *object_name, const char *fileName) {
if (cbd.status != S3StatusOK) { if (cbd.status != S3StatusOK) {
uError("%s: %d(%s)", __func__, cbd.status, cbd.err_msg); uError("%s: %d(%s)", __func__, cbd.status, cbd.err_msg);
taosCloseFile(&pFile); (void)taosCloseFile(&pFile);
TAOS_RETURN(TAOS_SYSTEM_ERROR(EIO)); TAOS_RETURN(TAOS_SYSTEM_ERROR(EIO));
} }
taosCloseFile(&pFile); (void)taosCloseFile(&pFile);
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
} }
@ -1250,9 +1250,9 @@ int32_t s3GetObjectsByPrefix(const char *prefix, const char *path) {
tmp = (tmp == NULL) ? object : tmp + 1; tmp = (tmp == NULL) ? object : tmp + 1;
char fileName[PATH_MAX] = {0}; char fileName[PATH_MAX] = {0};
if (path[strlen(path) - 1] != TD_DIRSEP_CHAR) { if (path[strlen(path) - 1] != TD_DIRSEP_CHAR) {
snprintf(fileName, PATH_MAX, "%s%s%s", path, TD_DIRSEP, tmp); (void)snprintf(fileName, PATH_MAX, "%s%s%s", path, TD_DIRSEP, tmp);
} else { } else {
snprintf(fileName, PATH_MAX, "%s%s", path, tmp); (void)snprintf(fileName, PATH_MAX, "%s%s", path, tmp);
} }
if (s3GetObjectToFile(object, fileName) != 0) { if (s3GetObjectToFile(object, fileName) != 0) {
taosArrayDestroyEx(objectArray, s3FreeObjectKey); taosArrayDestroyEx(objectArray, s3FreeObjectKey);

30
source/common/src/cos_cp.c
View File

@ -43,12 +43,12 @@ static int32_t cos_cp_parse_body(char* cp_body, SCheckpoint* cp) {
item = cJSON_GetObjectItem(json, "md5"); item = cJSON_GetObjectItem(json, "md5");
if (cJSON_IsString(item)) { if (cJSON_IsString(item)) {
memcpy(cp->md5, item->valuestring, strlen(item->valuestring)); (void)memcpy(cp->md5, item->valuestring, strlen(item->valuestring));
} }
item = cJSON_GetObjectItem(json, "upload_id"); item = cJSON_GetObjectItem(json, "upload_id");
if (cJSON_IsString(item)) { if (cJSON_IsString(item)) {
strncpy(cp->upload_id, item->valuestring, 128); (void)strncpy(cp->upload_id, item->valuestring, 128);
} }
item2 = cJSON_GetObjectItem(json, "file"); item2 = cJSON_GetObjectItem(json, "file");
@ -65,12 +65,12 @@ static int32_t cos_cp_parse_body(char* cp_body, SCheckpoint* cp) {
item = cJSON_GetObjectItem(item2, "path"); item = cJSON_GetObjectItem(item2, "path");
if (cJSON_IsString(item)) { if (cJSON_IsString(item)) {
strncpy(cp->file_path, item->valuestring, TSDB_FILENAME_LEN); (void)strncpy(cp->file_path, item->valuestring, TSDB_FILENAME_LEN);
} }
item = cJSON_GetObjectItem(item2, "file_md5"); item = cJSON_GetObjectItem(item2, "file_md5");
if (cJSON_IsString(item)) { if (cJSON_IsString(item)) {
strncpy(cp->file_md5, item->valuestring, 64); (void)strncpy(cp->file_md5, item->valuestring, 64);
} }
} }
@ -83,17 +83,17 @@ static int32_t cos_cp_parse_body(char* cp_body, SCheckpoint* cp) {
item = cJSON_GetObjectItem(item2, "object_name"); item = cJSON_GetObjectItem(item2, "object_name");
if (cJSON_IsString(item)) { if (cJSON_IsString(item)) {
strncpy(cp->object_name, item->valuestring, 128); (void)strncpy(cp->object_name, item->valuestring, 128);
} }
item = cJSON_GetObjectItem(item2, "object_last_modified"); item = cJSON_GetObjectItem(item2, "object_last_modified");
if (cJSON_IsString(item)) { if (cJSON_IsString(item)) {
strncpy(cp->object_last_modified, item->valuestring, 64); (void)strncpy(cp->object_last_modified, item->valuestring, 64);
} }
item = cJSON_GetObjectItem(item2, "object_etag"); item = cJSON_GetObjectItem(item2, "object_etag");
if (cJSON_IsString(item)) { if (cJSON_IsString(item)) {
strncpy(cp->object_etag, item->valuestring, 128); (void)strncpy(cp->object_etag, item->valuestring, 128);
} }
} }
@ -141,7 +141,7 @@ static int32_t cos_cp_parse_body(char* cp_body, SCheckpoint* cp) {
item3 = cJSON_GetObjectItem(item, "etag"); item3 = cJSON_GetObjectItem(item, "etag");
if (cJSON_IsString(item3)) { if (cJSON_IsString(item3)) {
strncpy(cp->parts[index].etag, item3->valuestring, 128); (void)strncpy(cp->parts[index].etag, item3->valuestring, 128);
} }
} }
} }
@ -179,7 +179,7 @@ int32_t cos_cp_load(char const* filepath, SCheckpoint* checkpoint) {
} else if (n != size) { } else if (n != size) {
TAOS_CHECK_GOTO(TSDB_CODE_FILE_CORRUPTED, &lino, _exit); TAOS_CHECK_GOTO(TSDB_CODE_FILE_CORRUPTED, &lino, _exit);
} }
taosCloseFile(&fd); (void)taosCloseFile(&fd);
cp_body[size] = '\0'; cp_body[size] = '\0';
return cos_cp_parse_body(cp_body, checkpoint); return cos_cp_parse_body(cp_body, checkpoint);
@ -189,7 +189,7 @@ _exit:
uError("%s failed at line %d since %s", __func__, lino, tstrerror(code)); uError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
} }
if (fd) { if (fd) {
taosCloseFile(&fd); (void)taosCloseFile(&fd);
} }
if (cp_body) { if (cp_body) {
taosMemoryFree(cp_body); taosMemoryFree(cp_body);
@ -309,7 +309,7 @@ int32_t cos_cp_dump(SCheckpoint* cp) {
if (!item) { if (!item) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit); TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
} }
cJSON_AddItemToArray(ajson, item); (void)cJSON_AddItemToArray(ajson, item);
if (NULL == cJSON_AddNumberToObject(item, "index", cp->parts[i].index)) { if (NULL == cJSON_AddNumberToObject(item, "index", cp->parts[i].index)) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit); TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
@ -346,7 +346,7 @@ void cos_cp_get_undo_parts(SCheckpoint* checkpoint, int* part_num, SCheckpointPa
void cos_cp_update(SCheckpoint* checkpoint, int32_t part_index, char const* etag, uint64_t crc64) { void cos_cp_update(SCheckpoint* checkpoint, int32_t part_index, char const* etag, uint64_t crc64) {
checkpoint->parts[part_index].completed = 1; checkpoint->parts[part_index].completed = 1;
strncpy(checkpoint->parts[part_index].etag, etag, 127); (void)strncpy(checkpoint->parts[part_index].etag, etag, 127);
checkpoint->parts[part_index].crc64 = crc64; checkpoint->parts[part_index].crc64 = crc64;
} }
@ -355,12 +355,12 @@ void cos_cp_build_upload(SCheckpoint* checkpoint, char const* filepath, int64_t
int i = 0; int i = 0;
checkpoint->cp_type = COS_CP_TYPE_UPLOAD; checkpoint->cp_type = COS_CP_TYPE_UPLOAD;
memset(checkpoint->file_path, 0, TSDB_FILENAME_LEN); (void)memset(checkpoint->file_path, 0, TSDB_FILENAME_LEN);
strncpy(checkpoint->file_path, filepath, TSDB_FILENAME_LEN - 1); (void)strncpy(checkpoint->file_path, filepath, TSDB_FILENAME_LEN - 1);
checkpoint->file_size = size; checkpoint->file_size = size;
checkpoint->file_last_modified = mtime; checkpoint->file_last_modified = mtime;
strncpy(checkpoint->upload_id, upload_id, 127); (void)strncpy(checkpoint->upload_id, upload_id, 127);
checkpoint->part_size = part_size; checkpoint->part_size = part_size;
for (; i * part_size < size; i++) { for (; i * part_size < size; i++) {

110
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -33,7 +33,7 @@ static int32_t tsdbOpenBCache(STsdb *pTsdb) {
taosLRUCacheSetStrictCapacity(pCache, false); taosLRUCacheSetStrictCapacity(pCache, false);
taosThreadMutexInit(&pTsdb->bMutex, NULL); (void)taosThreadMutexInit(&pTsdb->bMutex, NULL);
pTsdb->bCache = pCache; pTsdb->bCache = pCache;
@ -57,7 +57,7 @@ static void tsdbCloseBCache(STsdb *pTsdb) {
taosLRUCacheCleanup(pCache); taosLRUCacheCleanup(pCache);
taosThreadMutexDestroy(&pTsdb->bMutex); (void)taosThreadMutexDestroy(&pTsdb->bMutex);
} }
} }
@ -72,7 +72,7 @@ static int32_t tsdbOpenPgCache(STsdb *pTsdb) {
taosLRUCacheSetStrictCapacity(pCache, false); taosLRUCacheSetStrictCapacity(pCache, false);
taosThreadMutexInit(&pTsdb->pgMutex, NULL); (void)taosThreadMutexInit(&pTsdb->pgMutex, NULL);
pTsdb->pgCache = pCache; pTsdb->pgCache = pCache;
@ -95,7 +95,7 @@ static void tsdbClosePgCache(STsdb *pTsdb) {
taosLRUCacheCleanup(pCache); taosLRUCacheCleanup(pCache);
taosThreadMutexDestroy(&pTsdb->bMutex); (void)taosThreadMutexDestroy(&pTsdb->bMutex);
} }
} }
@ -225,7 +225,7 @@ static int32_t tsdbOpenRocksCache(STsdb *pTsdb) {
pTsdb->rCache.flushoptions = flushoptions; pTsdb->rCache.flushoptions = flushoptions;
pTsdb->rCache.db = db; pTsdb->rCache.db = db;
taosThreadMutexInit(&pTsdb->rCache.rMutex, NULL); (void)taosThreadMutexInit(&pTsdb->rCache.rMutex, NULL);
pTsdb->rCache.pTSchema = NULL; pTsdb->rCache.pTSchema = NULL;
@ -258,7 +258,7 @@ static void tsdbCloseRocksCache(STsdb *pTsdb) {
rocksdb_block_based_options_destroy(pTsdb->rCache.tableoptions); rocksdb_block_based_options_destroy(pTsdb->rCache.tableoptions);
rocksdb_cache_destroy(pTsdb->rCache.blockcache); rocksdb_cache_destroy(pTsdb->rCache.blockcache);
rocksdb_comparator_destroy(pTsdb->rCache.my_comparator); rocksdb_comparator_destroy(pTsdb->rCache.my_comparator);
taosThreadMutexDestroy(&pTsdb->rCache.rMutex); (void)taosThreadMutexDestroy(&pTsdb->rCache.rMutex);
taosMemoryFree(pTsdb->rCache.pTSchema); taosMemoryFree(pTsdb->rCache.pTSchema);
} }
@ -266,12 +266,12 @@ static void rocksMayWrite(STsdb *pTsdb, bool force, bool read, bool lock) {
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
if (read) { if (read) {
if (lock) { if (lock) {
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
} }
wb = pTsdb->rCache.rwritebatch; wb = pTsdb->rCache.rwritebatch;
} else { } else {
if (lock) { if (lock) {
taosThreadMutexLock(&pTsdb->rCache.rMutex); (void)taosThreadMutexLock(&pTsdb->rCache.rMutex);
} }
} }
@ -292,9 +292,9 @@ static void rocksMayWrite(STsdb *pTsdb, bool force, bool read, bool lock) {
if (lock) { if (lock) {
if (read) { if (read) {
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
} else { } else {
taosThreadMutexUnlock(&pTsdb->rCache.rMutex); (void)taosThreadMutexUnlock(&pTsdb->rCache.rMutex);
} }
} }
} }
@ -486,7 +486,7 @@ static void tsdbCachePutBatch(SLastCol *pLastCol, const void *key, size_t klen,
tsdbError("tsdb/cache: vgId:%d, serialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("tsdb/cache: vgId:%d, serialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
} }
taosThreadMutexLock(&rCache->rMutex); (void)taosThreadMutexLock(&rCache->rMutex);
rocksdb_writebatch_put(wb, (char *)key, klen, rocks_value, vlen); rocksdb_writebatch_put(wb, (char *)key, klen, rocks_value, vlen);
@ -507,7 +507,7 @@ static void tsdbCachePutBatch(SLastCol *pLastCol, const void *key, size_t klen,
state->flush_count = 0; state->flush_count = 0;
} }
taosThreadMutexUnlock(&rCache->rMutex); (void)taosThreadMutexUnlock(&rCache->rMutex);
} }
int tsdbCacheFlushDirty(const void *key, size_t klen, void *value, void *ud) { int tsdbCacheFlushDirty(const void *key, size_t klen, void *value, void *ud) {
@ -529,7 +529,7 @@ int32_t tsdbCacheCommit(STsdb *pTsdb) {
SLRUCache *pCache = pTsdb->lruCache; SLRUCache *pCache = pTsdb->lruCache;
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
taosLRUCacheApply(pCache, tsdbCacheFlushDirty, &pTsdb->flushState); taosLRUCacheApply(pCache, tsdbCacheFlushDirty, &pTsdb->flushState);
@ -537,7 +537,7 @@ int32_t tsdbCacheCommit(STsdb *pTsdb) {
rocksMayWrite(pTsdb, true, true, false); rocksMayWrite(pTsdb, true, true, false);
rocksdb_flush(pTsdb->rCache.db, pTsdb->rCache.flushoptions, &err); rocksdb_flush(pTsdb->rCache.db, pTsdb->rCache.flushoptions, &err);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
if (NULL != err) { if (NULL != err) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, err); tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, err);
@ -778,7 +778,7 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrapper *pSchemaRow) { int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrapper *pSchemaRow) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
if (suid < 0) { if (suid < 0) {
for (int i = 0; i < pSchemaRow->nCols; ++i) { for (int i = 0; i < pSchemaRow->nCols; ++i) {
@ -792,7 +792,7 @@ int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrap
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
code = metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, uid, -1, &pTSchema); code = metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, uid, -1, &pTSchema);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -808,7 +808,7 @@ int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrap
taosMemoryFree(pTSchema); taosMemoryFree(pTSchema);
} }
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -816,7 +816,7 @@ int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrap
int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrapper *pSchemaRow) { int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrapper *pSchemaRow) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheCommitNoLock(pTsdb); (void)tsdbCacheCommitNoLock(pTsdb);
@ -836,7 +836,7 @@ int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWra
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
code = metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, uid, -1, &pTSchema); code = metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, uid, -1, &pTSchema);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -858,7 +858,7 @@ int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWra
rocksMayWrite(pTsdb, true, false, false); rocksMayWrite(pTsdb, true, false, false);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -866,14 +866,14 @@ int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWra
int32_t tsdbCacheDropSubTables(STsdb *pTsdb, SArray *uids, tb_uid_t suid) { int32_t tsdbCacheDropSubTables(STsdb *pTsdb, SArray *uids, tb_uid_t suid) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheCommitNoLock(pTsdb); (void)tsdbCacheCommitNoLock(pTsdb);
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
code = metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, suid, -1, &pTSchema); code = metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, suid, -1, &pTSchema);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -899,7 +899,7 @@ int32_t tsdbCacheDropSubTables(STsdb *pTsdb, SArray *uids, tb_uid_t suid) {
rocksMayWrite(pTsdb, true, false, false); rocksMayWrite(pTsdb, true, false, false);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -907,13 +907,13 @@ int32_t tsdbCacheDropSubTables(STsdb *pTsdb, SArray *uids, tb_uid_t suid) {
int32_t tsdbCacheNewNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, int8_t col_type) { int32_t tsdbCacheNewNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, int8_t col_type) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0); (void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0);
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1); (void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1);
// rocksMayWrite(pTsdb, true, false, false); // rocksMayWrite(pTsdb, true, false, false);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
//(void)tsdbCacheCommit(pTsdb); //(void)tsdbCacheCommit(pTsdb);
TAOS_RETURN(code); TAOS_RETURN(code);
@ -922,7 +922,7 @@ int32_t tsdbCacheNewNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, int8_t
int32_t tsdbCacheDropNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, bool hasPrimayKey) { int32_t tsdbCacheDropNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, bool hasPrimayKey) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheCommitNoLock(pTsdb); (void)tsdbCacheCommitNoLock(pTsdb);
@ -930,7 +930,7 @@ int32_t tsdbCacheDropNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, bool h
rocksMayWrite(pTsdb, true, false, true); rocksMayWrite(pTsdb, true, false, true);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -938,7 +938,7 @@ int32_t tsdbCacheDropNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, bool h
int32_t tsdbCacheNewSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, int8_t col_type) { int32_t tsdbCacheNewSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, int8_t col_type) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
for (int i = 0; i < TARRAY_SIZE(uids); ++i) { for (int i = 0; i < TARRAY_SIZE(uids); ++i) {
tb_uid_t uid = ((tb_uid_t *)TARRAY_DATA(uids))[i]; tb_uid_t uid = ((tb_uid_t *)TARRAY_DATA(uids))[i];
@ -948,7 +948,7 @@ int32_t tsdbCacheNewSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, int8_t
} }
// rocksMayWrite(pTsdb, true, false, false); // rocksMayWrite(pTsdb, true, false, false);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
//(void)tsdbCacheCommit(pTsdb); //(void)tsdbCacheCommit(pTsdb);
TAOS_RETURN(code); TAOS_RETURN(code);
@ -957,7 +957,7 @@ int32_t tsdbCacheNewSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, int8_t
int32_t tsdbCacheDropSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, bool hasPrimayKey) { int32_t tsdbCacheDropSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, bool hasPrimayKey) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheCommitNoLock(pTsdb); (void)tsdbCacheCommitNoLock(pTsdb);
@ -969,7 +969,7 @@ int32_t tsdbCacheDropSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, bool
rocksMayWrite(pTsdb, true, false, true); rocksMayWrite(pTsdb, true, false, true);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -1045,7 +1045,7 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
SArray *remainCols = NULL; SArray *remainCols = NULL;
SLRUCache *pCache = pTsdb->lruCache; SLRUCache *pCache = pTsdb->lruCache;
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
for (int i = 0; i < num_keys; ++i) { for (int i = 0; i < num_keys; ++i) {
SLastUpdateCtx *updCtx = (SLastUpdateCtx *)taosArrayGet(updCtxArray, i); SLastUpdateCtx *updCtx = (SLastUpdateCtx *)taosArrayGet(updCtxArray, i);
@ -1088,7 +1088,7 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
keys_list_sizes = taosMemoryCalloc(num_keys, sizeof(size_t)); keys_list_sizes = taosMemoryCalloc(num_keys, sizeof(size_t));
if (!keys_list || !keys_list_sizes) { if (!keys_list || !keys_list_sizes) {
taosMemoryFree(keys_list); taosMemoryFree(keys_list);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY); TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
} }
for (int i = 0; i < num_keys; ++i) { for (int i = 0; i < num_keys; ++i) {
@ -1105,7 +1105,7 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
taosMemoryFree(keys_list_sizes); taosMemoryFree(keys_list_sizes);
taosMemoryFree(values_list); taosMemoryFree(values_list);
taosMemoryFree(values_list_sizes); taosMemoryFree(values_list_sizes);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY); TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
} }
rocksdb_multi_get(pTsdb->rCache.db, pTsdb->rCache.readoptions, num_keys, (const char *const *)keys_list, rocksdb_multi_get(pTsdb->rCache.db, pTsdb->rCache.readoptions, num_keys, (const char *const *)keys_list,
@ -1149,11 +1149,11 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
tsdbError("tsdb/cache: vgId:%d, serialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("tsdb/cache: vgId:%d, serialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
} }
taosThreadMutexLock(&pTsdb->rCache.rMutex); (void)taosThreadMutexLock(&pTsdb->rCache.rMutex);
rocksdb_writebatch_put(wb, (char *)&idxKey->key, ROCKS_KEY_LEN, value, vlen); rocksdb_writebatch_put(wb, (char *)&idxKey->key, ROCKS_KEY_LEN, value, vlen);
taosThreadMutexUnlock(&pTsdb->rCache.rMutex); (void)taosThreadMutexUnlock(&pTsdb->rCache.rMutex);
pLastCol = &lastColTmp; pLastCol = &lastColTmp;
SLastCol *pTmpLastCol = taosMemoryCalloc(1, sizeof(SLastCol)); SLastCol *pTmpLastCol = taosMemoryCalloc(1, sizeof(SLastCol));
@ -1164,7 +1164,7 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
taosMemoryFree(values_list_sizes); taosMemoryFree(values_list_sizes);
taosArrayDestroy(remainCols); taosArrayDestroy(remainCols);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY); TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
} }
*pTmpLastCol = *pLastCol; *pTmpLastCol = *pLastCol;
@ -1208,7 +1208,7 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
} }
_exit: _exit:
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
if (code) { if (code) {
tsdbError("tsdb/cache: vgId:%d, update failed at line %d since %s.", TD_VID(pTsdb->pVnode), lino, tstrerror(code)); tsdbError("tsdb/cache: vgId:%d, update failed at line %d since %s.", TD_VID(pTsdb->pVnode), lino, tstrerror(code));
@ -1699,7 +1699,7 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
} }
if (remainCols && TARRAY_SIZE(remainCols) > 0) { if (remainCols && TARRAY_SIZE(remainCols) > 0) {
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
for (int i = 0; i < TARRAY_SIZE(remainCols);) { for (int i = 0; i < TARRAY_SIZE(remainCols);) {
SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[i]; SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[i];
LRUHandle *h = taosLRUCacheLookup(pCache, &idxKey->key, ROCKS_KEY_LEN); LRUHandle *h = taosLRUCacheLookup(pCache, &idxKey->key, ROCKS_KEY_LEN);
@ -1710,13 +1710,13 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
for (int8_t j = 0; j < lastCol.rowKey.numOfPKs; j++) { for (int8_t j = 0; j < lastCol.rowKey.numOfPKs; j++) {
code = reallocVarDataVal(&lastCol.rowKey.pks[j]); code = reallocVarDataVal(&lastCol.rowKey.pks[j]);
if (code) { if (code) {
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
} }
code = reallocVarData(&lastCol.colVal); code = reallocVarData(&lastCol.colVal);
if (code) { if (code) {
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
taosArraySet(pLastArray, idxKey->idx, &lastCol); taosArraySet(pLastArray, idxKey->idx, &lastCol);
@ -1732,7 +1732,7 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
// tsdbTrace("tsdb/cache: vgId: %d, load %" PRId64 " from rocks", TD_VID(pTsdb->pVnode), uid); // tsdbTrace("tsdb/cache: vgId: %d, load %" PRId64 " from rocks", TD_VID(pTsdb->pVnode), uid);
code = tsdbCacheLoadFromRocks(pTsdb, uid, pLastArray, remainCols, pr, ltype); code = tsdbCacheLoadFromRocks(pTsdb, uid, pLastArray, remainCols, pr, ltype);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
if (remainCols) { if (remainCols) {
taosArrayDestroy(remainCols); taosArrayDestroy(remainCols);
@ -1792,13 +1792,13 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
(void)tsdbCacheCommit(pTsdb); (void)tsdbCacheCommit(pTsdb);
taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
taosThreadMutexLock(&pTsdb->rCache.rMutex); (void)taosThreadMutexLock(&pTsdb->rCache.rMutex);
// rocksMayWrite(pTsdb, true, false, false); // rocksMayWrite(pTsdb, true, false, false);
rocksdb_multi_get(pTsdb->rCache.db, pTsdb->rCache.readoptions, num_keys * 2, (const char *const *)keys_list, rocksdb_multi_get(pTsdb->rCache.db, pTsdb->rCache.readoptions, num_keys * 2, (const char *const *)keys_list,
keys_list_sizes, values_list, values_list_sizes, errs); keys_list_sizes, values_list, values_list_sizes, errs);
taosThreadMutexUnlock(&pTsdb->rCache.rMutex); (void)taosThreadMutexUnlock(&pTsdb->rCache.rMutex);
for (int i = 0; i < num_keys * 2; ++i) { for (int i = 0; i < num_keys * 2; ++i) {
if (errs[i]) { if (errs[i]) {
@ -1814,7 +1814,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
if (code) { if (code) {
tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
} }
taosThreadMutexLock(&pTsdb->rCache.rMutex); (void)taosThreadMutexLock(&pTsdb->rCache.rMutex);
if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) { if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) {
rocksdb_writebatch_delete(wb, keys_list[i], klen); rocksdb_writebatch_delete(wb, keys_list[i], klen);
} }
@ -1828,7 +1828,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) { if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) {
rocksdb_writebatch_delete(wb, keys_list[num_keys + i], klen); rocksdb_writebatch_delete(wb, keys_list[num_keys + i], klen);
} }
taosThreadMutexUnlock(&pTsdb->rCache.rMutex); (void)taosThreadMutexUnlock(&pTsdb->rCache.rMutex);
taosMemoryFreeClear(pLastCol); taosMemoryFreeClear(pLastCol);
rocksdb_free(values_list[i]); rocksdb_free(values_list[i]);
@ -1879,7 +1879,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
rocksMayWrite(pTsdb, true, false, true); rocksMayWrite(pTsdb, true, false, true);
taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
_exit: _exit:
taosMemoryFree(pTSchema); taosMemoryFree(pTSchema);
@ -1904,7 +1904,7 @@ int32_t tsdbOpenCache(STsdb *pTsdb) {
taosLRUCacheSetStrictCapacity(pCache, false); taosLRUCacheSetStrictCapacity(pCache, false);
taosThreadMutexInit(&pTsdb->lruMutex, NULL); (void)taosThreadMutexInit(&pTsdb->lruMutex, NULL);
pTsdb->flushState.pTsdb = pTsdb; pTsdb->flushState.pTsdb = pTsdb;
pTsdb->flushState.flush_count = 0; pTsdb->flushState.flush_count = 0;
@ -1926,7 +1926,7 @@ void tsdbCloseCache(STsdb *pTsdb) {
taosLRUCacheCleanup(pCache); taosLRUCacheCleanup(pCache);
taosThreadMutexDestroy(&pTsdb->lruMutex); (void)taosThreadMutexDestroy(&pTsdb->lruMutex);
} }
tsdbCloseBCache(pTsdb); tsdbCloseBCache(pTsdb);
@ -3365,7 +3365,7 @@ int32_t tsdbCacheGetBlockS3(SLRUCache *pCache, STsdbFD *pFD, LRUHandle **handle)
LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen); LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen);
if (!h) { if (!h) {
STsdb *pTsdb = pFD->pTsdb; STsdb *pTsdb = pFD->pTsdb;
taosThreadMutexLock(&pTsdb->bMutex); (void)taosThreadMutexLock(&pTsdb->bMutex);
h = taosLRUCacheLookup(pCache, key, keyLen); h = taosLRUCacheLookup(pCache, key, keyLen);
if (!h) { if (!h) {
@ -3373,7 +3373,7 @@ int32_t tsdbCacheGetBlockS3(SLRUCache *pCache, STsdbFD *pFD, LRUHandle **handle)
code = tsdbCacheLoadBlockS3(pFD, &pBlock); code = tsdbCacheLoadBlockS3(pFD, &pBlock);
// if table's empty or error, return code of -1 // if table's empty or error, return code of -1
if (code != TSDB_CODE_SUCCESS || pBlock == NULL) { if (code != TSDB_CODE_SUCCESS || pBlock == NULL) {
taosThreadMutexUnlock(&pTsdb->bMutex); (void)taosThreadMutexUnlock(&pTsdb->bMutex);
*handle = NULL; *handle = NULL;
if (code == TSDB_CODE_SUCCESS && !pBlock) { if (code == TSDB_CODE_SUCCESS && !pBlock) {
@ -3392,7 +3392,7 @@ int32_t tsdbCacheGetBlockS3(SLRUCache *pCache, STsdbFD *pFD, LRUHandle **handle)
} }
} }
taosThreadMutexUnlock(&pTsdb->bMutex); (void)taosThreadMutexUnlock(&pTsdb->bMutex);
} }
*handle = h; *handle = h;
@ -3418,7 +3418,7 @@ int32_t tsdbCacheSetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, uint8_
LRUHandle *handle = NULL; LRUHandle *handle = NULL;
getBCacheKey(pFD->fid, pFD->cid, pgno, key, &keyLen); getBCacheKey(pFD->fid, pFD->cid, pgno, key, &keyLen);
taosThreadMutexLock(&pFD->pTsdb->pgMutex); (void)taosThreadMutexLock(&pFD->pTsdb->pgMutex);
handle = taosLRUCacheLookup(pFD->pTsdb->pgCache, key, keyLen); handle = taosLRUCacheLookup(pFD->pTsdb->pgCache, key, keyLen);
if (!handle) { if (!handle) {
size_t charge = pFD->szPage; size_t charge = pFD->szPage;
@ -3436,7 +3436,7 @@ int32_t tsdbCacheSetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, uint8_
// code = TSDB_CODE_OUT_OF_MEMORY; // code = TSDB_CODE_OUT_OF_MEMORY;
} }
} }
taosThreadMutexUnlock(&pFD->pTsdb->pgMutex); (void)taosThreadMutexUnlock(&pFD->pTsdb->pgMutex);
tsdbCacheRelease(pFD->pTsdb->pgCache, handle); tsdbCacheRelease(pFD->pTsdb->pgCache, handle);

20
source/libs/sync/src/syncRaftLog.c
View File

@ -63,12 +63,12 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
pData->pWal = pSyncNode->pWal; pData->pWal = pSyncNode->pWal;
ASSERT(pData->pWal != NULL); ASSERT(pData->pWal != NULL);
taosThreadMutexInit(&(pData->mutex), NULL); (void)taosThreadMutexInit(&(pData->mutex), NULL);
pData->pWalHandle = walOpenReader(pData->pWal, NULL, 0); pData->pWalHandle = walOpenReader(pData->pWal, NULL, 0);
if (!pData->pWalHandle) { if (!pData->pWalHandle) {
taosMemoryFree(pLogStore); taosMemoryFree(pLogStore);
taosLRUCacheCleanup(pLogStore->pCache); taosLRUCacheCleanup(pLogStore->pCache);
taosThreadMutexDestroy(&(pData->mutex)); (void)taosThreadMutexDestroy(&(pData->mutex));
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL; return NULL;
} }
@ -96,13 +96,13 @@ void logStoreDestory(SSyncLogStore* pLogStore) {
if (pLogStore != NULL) { if (pLogStore != NULL) {
SSyncLogStoreData* pData = pLogStore->data; SSyncLogStoreData* pData = pLogStore->data;
taosThreadMutexLock(&(pData->mutex)); (void)taosThreadMutexLock(&(pData->mutex));
if (pData->pWalHandle != NULL) { if (pData->pWalHandle != NULL) {
walCloseReader(pData->pWalHandle); walCloseReader(pData->pWalHandle);
pData->pWalHandle = NULL; pData->pWalHandle = NULL;
} }
taosThreadMutexUnlock(&(pData->mutex)); (void)taosThreadMutexUnlock(&(pData->mutex));
taosThreadMutexDestroy(&(pData->mutex)); (void)taosThreadMutexDestroy(&(pData->mutex));
taosMemoryFree(pLogStore->data); taosMemoryFree(pLogStore->data);
@ -261,12 +261,12 @@ int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncR
*ppEntry = NULL; *ppEntry = NULL;
int64_t ts1 = taosGetTimestampNs(); int64_t ts1 = taosGetTimestampNs();
taosThreadMutexLock(&(pData->mutex)); (void)taosThreadMutexLock(&(pData->mutex));
SWalReader* pWalHandle = pData->pWalHandle; SWalReader* pWalHandle = pData->pWalHandle;
if (pWalHandle == NULL) { if (pWalHandle == NULL) {
sError("vgId:%d, wal handle is NULL", pData->pSyncNode->vgId); sError("vgId:%d, wal handle is NULL", pData->pSyncNode->vgId);
taosThreadMutexUnlock(&(pData->mutex)); (void)taosThreadMutexUnlock(&(pData->mutex));
TAOS_RETURN(TSDB_CODE_SYN_INTERNAL_ERROR); TAOS_RETURN(TSDB_CODE_SYN_INTERNAL_ERROR);
} }
@ -297,7 +297,7 @@ int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncR
terrno = saveErr; terrno = saveErr;
*/ */
taosThreadMutexUnlock(&(pData->mutex)); (void)taosThreadMutexUnlock(&(pData->mutex));
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -311,7 +311,7 @@ int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncR
(*ppEntry)->term = pWalHandle->pHead->head.syncMeta.term; (*ppEntry)->term = pWalHandle->pHead->head.syncMeta.term;
(*ppEntry)->index = index; (*ppEntry)->index = index;
ASSERT((*ppEntry)->dataLen == pWalHandle->pHead->head.bodyLen); ASSERT((*ppEntry)->dataLen == pWalHandle->pHead->head.bodyLen);
memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen); (void)memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);
/* /*
int32_t saveErr = terrno; int32_t saveErr = terrno;
@ -319,7 +319,7 @@ int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncR
terrno = saveErr; terrno = saveErr;
*/ */
taosThreadMutexUnlock(&(pData->mutex)); (void)taosThreadMutexUnlock(&(pData->mutex));
int64_t ts4 = taosGetTimestampNs(); int64_t ts4 = taosGetTimestampNs();
int64_t tsElapsed = ts4 - ts1; int64_t tsElapsed = ts4 - ts1;

28
source/libs/sync/src/syncRaftStore.c
View File

@ -152,52 +152,52 @@ _OVER:
} }
int32_t raftStoreOpen(SSyncNode *pNode) { int32_t raftStoreOpen(SSyncNode *pNode) {
taosThreadMutexInit(&pNode->raftStore.mutex, NULL); (void)taosThreadMutexInit(&pNode->raftStore.mutex, NULL);
return raftStoreReadFile(pNode); return raftStoreReadFile(pNode);
} }
void raftStoreClose(SSyncNode *pNode) { taosThreadMutexDestroy(&pNode->raftStore.mutex); } void raftStoreClose(SSyncNode *pNode) { (void)taosThreadMutexDestroy(&pNode->raftStore.mutex); }
bool raftStoreHasVoted(SSyncNode *pNode) { bool raftStoreHasVoted(SSyncNode *pNode) {
taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
bool b = syncUtilEmptyId(&pNode->raftStore.voteFor); bool b = syncUtilEmptyId(&pNode->raftStore.voteFor);
taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
return (!b); return (!b);
} }
void raftStoreVote(SSyncNode *pNode, SRaftId *pRaftId) { void raftStoreVote(SSyncNode *pNode, SRaftId *pRaftId) {
taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
pNode->raftStore.voteFor = *pRaftId; pNode->raftStore.voteFor = *pRaftId;
(void)raftStoreWriteFile(pNode); (void)raftStoreWriteFile(pNode);
taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
} }
void raftStoreClearVote(SSyncNode *pNode) { void raftStoreClearVote(SSyncNode *pNode) {
taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
pNode->raftStore.voteFor = EMPTY_RAFT_ID; pNode->raftStore.voteFor = EMPTY_RAFT_ID;
(void)raftStoreWriteFile(pNode); (void)raftStoreWriteFile(pNode);
taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
} }
void raftStoreNextTerm(SSyncNode *pNode) { void raftStoreNextTerm(SSyncNode *pNode) {
taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
pNode->raftStore.currentTerm++; pNode->raftStore.currentTerm++;
(void)raftStoreWriteFile(pNode); (void)raftStoreWriteFile(pNode);
taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
} }
void raftStoreSetTerm(SSyncNode *pNode, SyncTerm term) { void raftStoreSetTerm(SSyncNode *pNode, SyncTerm term) {
taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
if (pNode->raftStore.currentTerm < term) { if (pNode->raftStore.currentTerm < term) {
pNode->raftStore.currentTerm = term; pNode->raftStore.currentTerm = term;
(void)raftStoreWriteFile(pNode); (void)raftStoreWriteFile(pNode);
} }
taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
} }
SyncTerm raftStoreGetTerm(SSyncNode *pNode) { SyncTerm raftStoreGetTerm(SSyncNode *pNode) {
taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
SyncTerm term = pNode->raftStore.currentTerm; SyncTerm term = pNode->raftStore.currentTerm;
taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
return term; return term;
} }

8
source/libs/sync/src/syncReplication.c
View File

@ -48,19 +48,19 @@
int32_t syncNodeReplicateReset(SSyncNode* pNode, SRaftId* pDestId) { int32_t syncNodeReplicateReset(SSyncNode* pNode, SRaftId* pDestId) {
SSyncLogBuffer* pBuf = pNode->pLogBuf; SSyncLogBuffer* pBuf = pNode->pLogBuf;
taosThreadMutexLock(&pBuf->mutex); (void)taosThreadMutexLock(&pBuf->mutex);
SSyncLogReplMgr* pMgr = syncNodeGetLogReplMgr(pNode, pDestId); SSyncLogReplMgr* pMgr = syncNodeGetLogReplMgr(pNode, pDestId);
syncLogReplReset(pMgr); syncLogReplReset(pMgr);
taosThreadMutexUnlock(&pBuf->mutex); (void)taosThreadMutexUnlock(&pBuf->mutex);
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
} }
int32_t syncNodeReplicate(SSyncNode* pNode) { int32_t syncNodeReplicate(SSyncNode* pNode) {
SSyncLogBuffer* pBuf = pNode->pLogBuf; SSyncLogBuffer* pBuf = pNode->pLogBuf;
taosThreadMutexLock(&pBuf->mutex); (void)taosThreadMutexLock(&pBuf->mutex);
int32_t ret = syncNodeReplicateWithoutLock(pNode); int32_t ret = syncNodeReplicateWithoutLock(pNode);
taosThreadMutexUnlock(&pBuf->mutex); (void)taosThreadMutexUnlock(&pBuf->mutex);
TAOS_RETURN(ret); TAOS_RETURN(ret);
} }

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

@ -114,11 +114,11 @@ static inline SWalFileInfo* walGetCurFileInfo(SWal* pWal) {
} }
static inline void walBuildLogName(SWal* pWal, int64_t fileFirstVer, char* buf) { static inline void walBuildLogName(SWal* pWal, int64_t fileFirstVer, char* buf) {
sprintf(buf, "%s/%020" PRId64 "." WAL_LOG_SUFFIX, pWal->path, fileFirstVer); TAOS_UNUSED(sprintf(buf, "%s/%020" PRId64 "." WAL_LOG_SUFFIX, pWal->path, fileFirstVer));
} }
static inline void walBuildIdxName(SWal* pWal, int64_t fileFirstVer, char* buf) { static inline void walBuildIdxName(SWal* pWal, int64_t fileFirstVer, char* buf) {
sprintf(buf, "%s/%020" PRId64 "." WAL_INDEX_SUFFIX, pWal->path, fileFirstVer); TAOS_UNUSED(sprintf(buf, "%s/%020" PRId64 "." WAL_INDEX_SUFFIX, pWal->path, fileFirstVer));
} }
static inline int walValidHeadCksum(SWalCkHead* pHead) { static inline int walValidHeadCksum(SWalCkHead* pHead) {

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

@ -57,7 +57,7 @@ static FORCE_INLINE int32_t walScanLogGetLastVer(SWal* pWal, int32_t fileIdx, in
walBuildLogName(pWal, pFileInfo->firstVer, fnameStr); walBuildLogName(pWal, pFileInfo->firstVer, fnameStr);
int64_t fileSize = 0; int64_t fileSize = 0;
taosStatFile(fnameStr, &fileSize, NULL, NULL); (void)taosStatFile(fnameStr, &fileSize, NULL, NULL);
TdFilePtr pFile = taosOpenFile(fnameStr, TD_FILE_READ | TD_FILE_WRITE); TdFilePtr pFile = taosOpenFile(fnameStr, TD_FILE_READ | TD_FILE_WRITE);
if (pFile == NULL) { if (pFile == NULL) {
@ -362,7 +362,7 @@ static int32_t walTrimIdxFile(SWal* pWal, int32_t fileIdx) {
walBuildIdxName(pWal, pFileInfo->firstVer, fnameStr); walBuildIdxName(pWal, pFileInfo->firstVer, fnameStr);
int64_t fileSize = 0; int64_t fileSize = 0;
taosStatFile(fnameStr, &fileSize, NULL, NULL); (void)taosStatFile(fnameStr, &fileSize, NULL, NULL);
int64_t records = TMAX(0, pFileInfo->lastVer - pFileInfo->firstVer + 1); int64_t records = TMAX(0, pFileInfo->lastVer - pFileInfo->firstVer + 1);
int64_t lastEndOffset = records * sizeof(SWalIdxEntry); int64_t lastEndOffset = records * sizeof(SWalIdxEntry);
@ -378,8 +378,8 @@ static int32_t walTrimIdxFile(SWal* pWal, int32_t fileIdx) {
wInfo("vgId:%d, trim idx file. file: %s, size: %" PRId64 ", offset: %" PRId64, pWal->cfg.vgId, fnameStr, fileSize, wInfo("vgId:%d, trim idx file. file: %s, size: %" PRId64 ", offset: %" PRId64, pWal->cfg.vgId, fnameStr, fileSize,
lastEndOffset); lastEndOffset);
taosFtruncateFile(pFile, lastEndOffset); (void)taosFtruncateFile(pFile, lastEndOffset);
taosCloseFile(&pFile); (void)taosCloseFile(&pFile);
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
} }
@ -392,8 +392,8 @@ int32_t walCheckAndRepairMeta(SWal* pWal) {
regex_t logRegPattern; regex_t logRegPattern;
regex_t idxRegPattern; regex_t idxRegPattern;
regcomp(&logRegPattern, logPattern, REG_EXTENDED); (void)regcomp(&logRegPattern, logPattern, REG_EXTENDED);
regcomp(&idxRegPattern, idxPattern, REG_EXTENDED); (void)regcomp(&idxRegPattern, idxPattern, REG_EXTENDED);
TdDirPtr pDir = taosOpenDir(pWal->path); TdDirPtr pDir = taosOpenDir(pWal->path);
if (pDir == NULL) { if (pDir == NULL) {
@ -412,13 +412,13 @@ int32_t walCheckAndRepairMeta(SWal* pWal) {
int code = regexec(&logRegPattern, name, 0, NULL, 0); int code = regexec(&logRegPattern, name, 0, NULL, 0);
if (code == 0) { if (code == 0) {
SWalFileInfo fileInfo; SWalFileInfo fileInfo;
memset(&fileInfo, -1, sizeof(SWalFileInfo)); (void)memset(&fileInfo, -1, sizeof(SWalFileInfo));
sscanf(name, "%" PRId64 ".log", &fileInfo.firstVer); (void)sscanf(name, "%" PRId64 ".log", &fileInfo.firstVer);
taosArrayPush(actualLog, &fileInfo); (void)taosArrayPush(actualLog, &fileInfo);
} }
} }
taosCloseDir(&pDir); (void)taosCloseDir(&pDir);
regfree(&logRegPattern); regfree(&logRegPattern);
regfree(&idxRegPattern); regfree(&idxRegPattern);
@ -549,7 +549,7 @@ static int32_t walCheckAndRepairIdxFile(SWal* pWal, int32_t fileIdx) {
TdFilePtr pIdxFile = NULL; TdFilePtr pIdxFile = NULL;
SWalIdxEntry idxEntry = {.ver = pFileInfo->firstVer - 1, .offset = -sizeof(SWalCkHead)}; SWalIdxEntry idxEntry = {.ver = pFileInfo->firstVer - 1, .offset = -sizeof(SWalCkHead)};
SWalCkHead ckHead; SWalCkHead ckHead;
memset(&ckHead, 0, sizeof(ckHead)); (void)memset(&ckHead, 0, sizeof(ckHead));
ckHead.head.version = idxEntry.ver; ckHead.head.version = idxEntry.ver;
pIdxFile = taosOpenFile(fnameStr, TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE); pIdxFile = taosOpenFile(fnameStr, TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE);
@ -675,7 +675,7 @@ _err:
int64_t walGetVerRetention(SWal* pWal, int64_t bytes) { int64_t walGetVerRetention(SWal* pWal, int64_t bytes) {
int64_t ver = -1; int64_t ver = -1;
int64_t totSize = 0; int64_t totSize = 0;
taosThreadMutexLock(&pWal->mutex); (void)taosThreadMutexLock(&pWal->mutex);
int32_t fileIdx = taosArrayGetSize(pWal->fileInfoSet); int32_t fileIdx = taosArrayGetSize(pWal->fileInfoSet);
while (--fileIdx) { while (--fileIdx) {
SWalFileInfo* pInfo = taosArrayGet(pWal->fileInfoSet, fileIdx); SWalFileInfo* pInfo = taosArrayGet(pWal->fileInfoSet, fileIdx);
@ -685,7 +685,7 @@ int64_t walGetVerRetention(SWal* pWal, int64_t bytes) {
} }
totSize += pInfo->fileSize; totSize += pInfo->fileSize;
} }
taosThreadMutexUnlock(&pWal->mutex); (void)taosThreadMutexUnlock(&pWal->mutex);
return ver + 1; return ver + 1;
} }
@ -727,7 +727,7 @@ int32_t walRollFileInfo(SWal* pWal) {
pNewInfo->closeTs = -1; pNewInfo->closeTs = -1;
pNewInfo->fileSize = 0; pNewInfo->fileSize = 0;
pNewInfo->syncedOffset = 0; pNewInfo->syncedOffset = 0;
taosArrayPush(pArray, pNewInfo); (void)taosArrayPush(pArray, pNewInfo);
taosMemoryFree(pNewInfo); taosMemoryFree(pNewInfo);
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
@ -753,21 +753,21 @@ int32_t walMetaSerialize(SWal* pWal, char** serialized) {
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY); TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
} }
cJSON_AddItemToObject(pRoot, "meta", pMeta); (void)cJSON_AddItemToObject(pRoot, "meta", pMeta);
sprintf(buf, "%" PRId64, pWal->vers.firstVer); (void)sprintf(buf, "%" PRId64, pWal->vers.firstVer);
cJSON_AddStringToObject(pMeta, "firstVer", buf); (void)cJSON_AddStringToObject(pMeta, "firstVer", buf);
sprintf(buf, "%" PRId64, pWal->vers.snapshotVer); (void)sprintf(buf, "%" PRId64, pWal->vers.snapshotVer);
cJSON_AddStringToObject(pMeta, "snapshotVer", buf); (void)cJSON_AddStringToObject(pMeta, "snapshotVer", buf);
sprintf(buf, "%" PRId64, pWal->vers.commitVer); (void)sprintf(buf, "%" PRId64, pWal->vers.commitVer);
cJSON_AddStringToObject(pMeta, "commitVer", buf); (void)cJSON_AddStringToObject(pMeta, "commitVer", buf);
sprintf(buf, "%" PRId64, pWal->vers.lastVer); (void)sprintf(buf, "%" PRId64, pWal->vers.lastVer);
cJSON_AddStringToObject(pMeta, "lastVer", buf); (void)cJSON_AddStringToObject(pMeta, "lastVer", buf);
cJSON_AddItemToObject(pRoot, "files", pFiles); (void)cJSON_AddItemToObject(pRoot, "files", pFiles);
SWalFileInfo* pData = pWal->fileInfoSet->pData; SWalFileInfo* pData = pWal->fileInfoSet->pData;
for (int i = 0; i < sz; i++) { for (int i = 0; i < sz; i++) {
SWalFileInfo* pInfo = &pData[i]; SWalFileInfo* pInfo = &pData[i];
cJSON_AddItemToArray(pFiles, pField = cJSON_CreateObject()); (void)cJSON_AddItemToArray(pFiles, pField = cJSON_CreateObject());
if (pField == NULL) { if (pField == NULL) {
cJSON_Delete(pRoot); cJSON_Delete(pRoot);
@ -775,16 +775,16 @@ int32_t walMetaSerialize(SWal* pWal, char** serialized) {
} }
// 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, "%" PRId64, pInfo->firstVer); (void)sprintf(buf, "%" PRId64, pInfo->firstVer);
cJSON_AddStringToObject(pField, "firstVer", buf); (void)cJSON_AddStringToObject(pField, "firstVer", buf);
sprintf(buf, "%" PRId64, pInfo->lastVer); (void)sprintf(buf, "%" PRId64, pInfo->lastVer);
cJSON_AddStringToObject(pField, "lastVer", buf); (void)cJSON_AddStringToObject(pField, "lastVer", buf);
sprintf(buf, "%" PRId64, pInfo->createTs); (void)sprintf(buf, "%" PRId64, pInfo->createTs);
cJSON_AddStringToObject(pField, "createTs", buf); (void)cJSON_AddStringToObject(pField, "createTs", buf);
sprintf(buf, "%" PRId64, pInfo->closeTs); (void)sprintf(buf, "%" PRId64, pInfo->closeTs);
cJSON_AddStringToObject(pField, "closeTs", buf); (void)cJSON_AddStringToObject(pField, "closeTs", buf);
sprintf(buf, "%" PRId64, pInfo->fileSize); (void)sprintf(buf, "%" PRId64, pInfo->fileSize);
cJSON_AddStringToObject(pField, "fileSize", buf); (void)cJSON_AddStringToObject(pField, "fileSize", buf);
} }
char* pSerialized = cJSON_Print(pRoot); char* pSerialized = cJSON_Print(pRoot);
cJSON_Delete(pRoot); cJSON_Delete(pRoot);
@ -818,7 +818,7 @@ int32_t walMetaDeserialize(SWal* pWal, const char* bytes) {
int sz = cJSON_GetArraySize(pFiles); int sz = cJSON_GetArraySize(pFiles);
// deserialize // deserialize
SArray* pArray = pWal->fileInfoSet; SArray* pArray = pWal->fileInfoSet;
taosArrayEnsureCap(pArray, sz); (void)taosArrayEnsureCap(pArray, sz);
for (int i = 0; i < sz; i++) { for (int i = 0; i < sz; i++) {
pInfoJson = cJSON_GetArrayItem(pFiles, i); pInfoJson = cJSON_GetArrayItem(pFiles, i);
@ -841,7 +841,7 @@ int32_t walMetaDeserialize(SWal* pWal, const char* bytes) {
pField = cJSON_GetObjectItem(pInfoJson, "fileSize"); pField = cJSON_GetObjectItem(pInfoJson, "fileSize");
if (!pField) goto _err; if (!pField) goto _err;
info.fileSize = atoll(cJSON_GetStringValue(pField)); info.fileSize = atoll(cJSON_GetStringValue(pField));
taosArrayPush(pArray, &info); (void)taosArrayPush(pArray, &info);
} }
pWal->fileInfoSet = pArray; pWal->fileInfoSet = pArray;
pWal->writeCur = sz - 1; pWal->writeCur = sz - 1;
@ -856,7 +856,7 @@ _err:
static int walFindCurMetaVer(SWal* pWal) { static int walFindCurMetaVer(SWal* pWal) {
const char* pattern = "^meta-ver[0-9]+$"; const char* pattern = "^meta-ver[0-9]+$";
regex_t walMetaRegexPattern; regex_t walMetaRegexPattern;
regcomp(&walMetaRegexPattern, pattern, REG_EXTENDED); (void)regcomp(&walMetaRegexPattern, pattern, REG_EXTENDED);
TdDirPtr pDir = taosOpenDir(pWal->path); TdDirPtr pDir = taosOpenDir(pWal->path);
if (pDir == NULL) { if (pDir == NULL) {
@ -872,13 +872,13 @@ static int walFindCurMetaVer(SWal* pWal) {
char* name = taosDirEntryBaseName(taosGetDirEntryName(pDirEntry)); char* name = taosDirEntryBaseName(taosGetDirEntryName(pDirEntry));
int code = regexec(&walMetaRegexPattern, name, 0, NULL, 0); int code = regexec(&walMetaRegexPattern, name, 0, NULL, 0);
if (code == 0) { if (code == 0) {
sscanf(name, "meta-ver%d", &metaVer); (void)sscanf(name, "meta-ver%d", &metaVer);
wDebug("vgId:%d, wal find current meta: %s is the meta file, ver %d", pWal->cfg.vgId, name, metaVer); wDebug("vgId:%d, wal find current meta: %s is the meta file, ver %d", pWal->cfg.vgId, name, metaVer);
break; break;
} }
wDebug("vgId:%d, wal find current meta: %s is not meta file", pWal->cfg.vgId, name); wDebug("vgId:%d, wal find current meta: %s is not meta file", pWal->cfg.vgId, name);
} }
taosCloseDir(&pDir); (void)taosCloseDir(&pDir);
regfree(&walMetaRegexPattern); regfree(&walMetaRegexPattern);
return metaVer; return metaVer;
} }
@ -961,8 +961,8 @@ int32_t walSaveMeta(SWal* pWal) {
// delete old file // delete old file
if (metaVer > -1) { if (metaVer > -1) {
walBuildMetaName(pWal, metaVer, fnameStr); (void)walBuildMetaName(pWal, metaVer, fnameStr);
taosRemoveFile(fnameStr); (void)taosRemoveFile(fnameStr);
} }
taosMemoryFree(serialized); taosMemoryFree(serialized);
@ -984,10 +984,10 @@ int32_t walLoadMeta(SWal* pWal) {
TAOS_RETURN(TSDB_CODE_FAILED); TAOS_RETURN(TSDB_CODE_FAILED);
} }
char fnameStr[WAL_FILE_LEN]; char fnameStr[WAL_FILE_LEN];
walBuildMetaName(pWal, metaVer, fnameStr); (void)walBuildMetaName(pWal, metaVer, fnameStr);
// read metafile // read metafile
int64_t fileSize = 0; int64_t fileSize = 0;
taosStatFile(fnameStr, &fileSize, NULL, NULL); (void)taosStatFile(fnameStr, &fileSize, NULL, NULL);
if (fileSize == 0) { if (fileSize == 0) {
(void)taosRemoveFile(fnameStr); (void)taosRemoveFile(fnameStr);
wDebug("vgId:%d, wal find empty meta ver %d", pWal->cfg.vgId, metaVer); wDebug("vgId:%d, wal find empty meta ver %d", pWal->cfg.vgId, metaVer);
@ -999,7 +999,7 @@ int32_t walLoadMeta(SWal* pWal) {
if (buf == NULL) { if (buf == NULL) {
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY); TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
} }
memset(buf, 0, size + 5); (void)memset(buf, 0, size + 5);
TdFilePtr pFile = taosOpenFile(fnameStr, TD_FILE_READ); TdFilePtr pFile = taosOpenFile(fnameStr, TD_FILE_READ);
if (pFile == NULL) { if (pFile == NULL) {
taosMemoryFree(buf); taosMemoryFree(buf);
@ -1007,7 +1007,7 @@ int32_t walLoadMeta(SWal* pWal) {
TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED); TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED);
} }
if (taosReadFile(pFile, buf, size) != size) { if (taosReadFile(pFile, buf, size) != size) {
taosCloseFile(&pFile); (void)taosCloseFile(&pFile);
taosMemoryFree(buf); taosMemoryFree(buf);
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
@ -1018,7 +1018,7 @@ int32_t walLoadMeta(SWal* pWal) {
wError("failed to deserialize wal meta. file:%s", fnameStr); wError("failed to deserialize wal meta. file:%s", fnameStr);
code = TSDB_CODE_WAL_FILE_CORRUPTED; code = TSDB_CODE_WAL_FILE_CORRUPTED;
} }
taosCloseFile(&pFile); (void)taosCloseFile(&pFile);
taosMemoryFree(buf); taosMemoryFree(buf);
TAOS_RETURN(code); TAOS_RETURN(code);
@ -1028,6 +1028,6 @@ int32_t walRemoveMeta(SWal* pWal) {
int metaVer = walFindCurMetaVer(pWal); int metaVer = walFindCurMetaVer(pWal);
if (metaVer == -1) return 0; if (metaVer == -1) return 0;
char fnameStr[WAL_FILE_LEN]; char fnameStr[WAL_FILE_LEN];
walBuildMetaName(pWal, metaVer, fnameStr); (void)walBuildMetaName(pWal, metaVer, fnameStr);
return taosRemoveFile(fnameStr); return taosRemoveFile(fnameStr);
} }

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

@ -33,7 +33,7 @@ static int32_t walCreateThread();
static void walStopThread(); static void walStopThread();
static void walFreeObj(void *pWal); static void walFreeObj(void *pWal);
int64_t walGetSeq() { return (int64_t)atomic_load_32(&tsWal.seq); } int64_t walGetSeq() { return (int64_t)atomic_load_32((volatile int32_t *)&tsWal.seq); }
int32_t walInit() { int32_t walInit() {
int8_t old; int8_t old;
@ -69,7 +69,7 @@ void walCleanUp() {
if (old == 1) { if (old == 1) {
walStopThread(); walStopThread();
taosCloseRef(tsWal.refSetId); TAOS_UNUSED(taosCloseRef(tsWal.refSetId));
wInfo("wal module is cleaned up"); wInfo("wal module is cleaned up");
atomic_store_8(&tsWal.inited, 0); atomic_store_8(&tsWal.inited, 0);
} }
@ -89,7 +89,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
} }
// set config // set config
memcpy(&pWal->cfg, pCfg, sizeof(SWalCfg)); TAOS_UNUSED(memcpy(&pWal->cfg, pCfg, sizeof(SWalCfg)));
pWal->fsyncSeq = pCfg->fsyncPeriod / 1000; pWal->fsyncSeq = pCfg->fsyncPeriod / 1000;
if (pWal->cfg.retentionSize > 0) { if (pWal->cfg.retentionSize > 0) {
@ -140,7 +140,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
pWal->lastRollSeq = -1; pWal->lastRollSeq = -1;
// init write buffer // init write buffer
memset(&pWal->writeHead, 0, sizeof(SWalCkHead)); TAOS_UNUSED(memset(&pWal->writeHead, 0, sizeof(SWalCkHead)));
pWal->writeHead.head.protoVer = WAL_PROTO_VER; pWal->writeHead.head.protoVer = WAL_PROTO_VER;
pWal->writeHead.magic = WAL_MAGIC; pWal->writeHead.magic = WAL_MAGIC;
@ -171,7 +171,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
_err: _err:
taosArrayDestroy(pWal->fileInfoSet); taosArrayDestroy(pWal->fileInfoSet);
taosHashCleanup(pWal->pRefHash); taosHashCleanup(pWal->pRefHash);
taosThreadMutexDestroy(&pWal->mutex); TAOS_UNUSED(taosThreadMutexDestroy(&pWal->mutex));
taosMemoryFreeClear(pWal); taosMemoryFreeClear(pWal);
return NULL; return NULL;
@ -207,19 +207,19 @@ int32_t walAlter(SWal *pWal, SWalCfg *pCfg) {
int32_t walPersist(SWal *pWal) { int32_t walPersist(SWal *pWal) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
code = walSaveMeta(pWal); code = walSaveMeta(pWal);
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(code); TAOS_RETURN(code);
} }
void walClose(SWal *pWal) { void walClose(SWal *pWal) {
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
(void)walSaveMeta(pWal); (void)walSaveMeta(pWal);
taosCloseFile(&pWal->pLogFile); TAOS_UNUSED(taosCloseFile(&pWal->pLogFile));
pWal->pLogFile = NULL; pWal->pLogFile = NULL;
taosCloseFile(&pWal->pIdxFile); (void)taosCloseFile(&pWal->pIdxFile);
pWal->pIdxFile = NULL; pWal->pIdxFile = NULL;
taosArrayDestroy(pWal->fileInfoSet); taosArrayDestroy(pWal->fileInfoSet);
pWal->fileInfoSet = NULL; pWal->fileInfoSet = NULL;
@ -235,22 +235,22 @@ void walClose(SWal *pWal) {
} }
taosHashCleanup(pWal->pRefHash); taosHashCleanup(pWal->pRefHash);
pWal->pRefHash = NULL; pWal->pRefHash = NULL;
taosThreadMutexUnlock(&pWal->mutex); (void)taosThreadMutexUnlock(&pWal->mutex);
if (pWal->cfg.level == TAOS_WAL_SKIP) { if (pWal->cfg.level == TAOS_WAL_SKIP) {
wInfo("vgId:%d, remove all wals, path:%s", pWal->cfg.vgId, pWal->path); wInfo("vgId:%d, remove all wals, path:%s", pWal->cfg.vgId, pWal->path);
taosRemoveDir(pWal->path); taosRemoveDir(pWal->path);
taosMkDir(pWal->path); (void)taosMkDir(pWal->path);
} }
taosRemoveRef(tsWal.refSetId, pWal->refId); (void)taosRemoveRef(tsWal.refSetId, pWal->refId);
} }
static void walFreeObj(void *wal) { static void walFreeObj(void *wal) {
SWal *pWal = wal; SWal *pWal = wal;
wDebug("vgId:%d, wal:%p is freed", pWal->cfg.vgId, pWal); wDebug("vgId:%d, wal:%p is freed", pWal->cfg.vgId, pWal);
taosThreadMutexDestroy(&pWal->mutex); (void)taosThreadMutexDestroy(&pWal->mutex);
taosMemoryFreeClear(pWal); taosMemoryFreeClear(pWal);
} }
@ -259,7 +259,7 @@ static bool walNeedFsync(SWal *pWal) {
return false; return false;
} }
if (atomic_load_32(&tsWal.seq) % pWal->fsyncSeq == 0) { if (atomic_load_32((volatile int32_t *)&tsWal.seq) % pWal->fsyncSeq == 0) {
return true; return true;
} }
@ -268,7 +268,7 @@ static bool walNeedFsync(SWal *pWal) {
static void walUpdateSeq() { static void walUpdateSeq() {
taosMsleep(WAL_REFRESH_MS); taosMsleep(WAL_REFRESH_MS);
atomic_add_fetch_32(&tsWal.seq, 1); (void)atomic_add_fetch_32((volatile int32_t *)&tsWal.seq, 1);
} }
static void walFsyncAll() { static void walFsyncAll() {
@ -276,7 +276,7 @@ static void walFsyncAll() {
while (pWal) { while (pWal) {
if (walNeedFsync(pWal)) { if (walNeedFsync(pWal)) {
wTrace("vgId:%d, do fsync, level:%d seq:%d rseq:%d", pWal->cfg.vgId, pWal->cfg.level, pWal->fsyncSeq, wTrace("vgId:%d, do fsync, level:%d seq:%d rseq:%d", pWal->cfg.vgId, pWal->cfg.level, pWal->fsyncSeq,
atomic_load_32(&tsWal.seq)); atomic_load_32((volatile int32_t *)&tsWal.seq));
int32_t code = taosFsyncFile(pWal->pLogFile); int32_t code = taosFsyncFile(pWal->pLogFile);
if (code != 0) { if (code != 0) {
wError("vgId:%d, file:%" PRId64 ".log, failed to fsync since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal), wError("vgId:%d, file:%" PRId64 ".log, failed to fsync since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
@ -301,8 +301,8 @@ static void *walThreadFunc(void *param) {
static int32_t walCreateThread() { static int32_t walCreateThread() {
TdThreadAttr thAttr; TdThreadAttr thAttr;
taosThreadAttrInit(&thAttr); (void)taosThreadAttrInit(&thAttr);
taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE); (void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
if (taosThreadCreate(&tsWal.thread, &thAttr, walThreadFunc, NULL) != 0) { if (taosThreadCreate(&tsWal.thread, &thAttr, walThreadFunc, NULL) != 0) {
wError("failed to create wal thread since %s", strerror(errno)); wError("failed to create wal thread since %s", strerror(errno));
@ -310,7 +310,7 @@ static int32_t walCreateThread() {
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
taosThreadAttrDestroy(&thAttr); (void)taosThreadAttrDestroy(&thAttr);
wDebug("wal thread is launched, thread:0x%08" PRIx64, taosGetPthreadId(tsWal.thread)); wDebug("wal thread is launched, thread:0x%08" PRIx64, taosGetPthreadId(tsWal.thread));
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
@ -320,7 +320,7 @@ static void walStopThread() {
atomic_store_8(&tsWal.stop, 1); atomic_store_8(&tsWal.stop, 1);
if (taosCheckPthreadValid(tsWal.thread)) { if (taosCheckPthreadValid(tsWal.thread)) {
taosThreadJoin(tsWal.thread, NULL); (void)taosThreadJoin(tsWal.thread, NULL);
taosThreadClear(&tsWal.thread); taosThreadClear(&tsWal.thread);
} }

48
source/libs/wal/src/walRead.c
View File

@ -41,7 +41,7 @@ SWalReader *walOpenReader(SWal *pWal, SWalFilterCond *cond, int64_t id) {
pReader->cond.enableRef = 0; pReader->cond.enableRef = 0;
} }
taosThreadMutexInit(&pReader->mutex, NULL); TAOS_UNUSED(taosThreadMutexInit(&pReader->mutex, NULL));
pReader->pHead = taosMemoryMalloc(sizeof(SWalCkHead)); pReader->pHead = taosMemoryMalloc(sizeof(SWalCkHead));
if (pReader->pHead == NULL) { if (pReader->pHead == NULL) {
@ -60,8 +60,8 @@ SWalReader *walOpenReader(SWal *pWal, SWalFilterCond *cond, int64_t id) {
void walCloseReader(SWalReader *pReader) { void walCloseReader(SWalReader *pReader) {
if (pReader == NULL) return; if (pReader == NULL) return;
taosCloseFile(&pReader->pIdxFile); TAOS_UNUSED(taosCloseFile(&pReader->pIdxFile));
taosCloseFile(&pReader->pLogFile); TAOS_UNUSED(taosCloseFile(&pReader->pLogFile));
taosMemoryFreeClear(pReader->pHead); taosMemoryFreeClear(pReader->pHead);
taosMemoryFree(pReader); taosMemoryFree(pReader);
} }
@ -115,9 +115,9 @@ void walReaderValidVersionRange(SWalReader *pReader, int64_t *sver, int64_t *eve
void walReaderVerifyOffset(SWalReader *pWalReader, STqOffsetVal *pOffset) { void walReaderVerifyOffset(SWalReader *pWalReader, STqOffsetVal *pOffset) {
// if offset version is small than first version , let's seek to first version // if offset version is small than first version , let's seek to first version
taosThreadMutexLock(&pWalReader->pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWalReader->pWal->mutex));
int64_t firstVer = walGetFirstVer((pWalReader)->pWal); int64_t firstVer = walGetFirstVer((pWalReader)->pWal);
taosThreadMutexUnlock(&pWalReader->pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWalReader->pWal->mutex));
if (pOffset->version < firstVer) { if (pOffset->version < firstVer) {
pOffset->version = firstVer; pOffset->version = firstVer;
@ -167,8 +167,8 @@ static int32_t walReadSeekFilePos(SWalReader *pReader, int64_t fileFirstVer, int
static int32_t walReadChangeFile(SWalReader *pReader, int64_t fileFirstVer) { static int32_t walReadChangeFile(SWalReader *pReader, int64_t fileFirstVer) {
char fnameStr[WAL_FILE_LEN] = {0}; char fnameStr[WAL_FILE_LEN] = {0};
taosCloseFile(&pReader->pIdxFile); TAOS_UNUSED(taosCloseFile(&pReader->pIdxFile));
taosCloseFile(&pReader->pLogFile); TAOS_UNUSED(taosCloseFile(&pReader->pLogFile));
walBuildLogName(pReader->pWal, fileFirstVer, fnameStr); walBuildLogName(pReader->pWal, fileFirstVer, fnameStr);
TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_READ); TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_READ);
@ -393,13 +393,13 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
TAOS_RETURN(TSDB_CODE_WAL_LOG_NOT_EXIST); TAOS_RETURN(TSDB_CODE_WAL_LOG_NOT_EXIST);
} }
taosThreadMutexLock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexLock(&pReader->mutex));
if (pReader->curVersion != ver) { if (pReader->curVersion != ver) {
code = walReaderSeekVer(pReader, ver); code = walReaderSeekVer(pReader, ver);
if (code) { if (code) {
wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since %s", pReader->pWal->cfg.vgId, ver, terrstr()); wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since %s", pReader->pWal->cfg.vgId, ver, terrstr());
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -414,7 +414,7 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
} else if (contLen == 0 && !seeked) { } else if (contLen == 0 && !seeked) {
code = walReadSeekVerImpl(pReader, ver); code = walReadSeekVerImpl(pReader, ver);
if (code) { if (code) {
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -423,7 +423,7 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
} else { } else {
wError("vgId:%d, failed to read WAL record head, index:%" PRId64 ", from log file since %s", wError("vgId:%d, failed to read WAL record head, index:%" PRId64 ", from log file since %s",
pReader->pWal->cfg.vgId, ver, terrstr()); pReader->pWal->cfg.vgId, ver, terrstr());
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
if (contLen < 0) { if (contLen < 0) {
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
@ -437,7 +437,7 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
if (code != 0) { if (code != 0) {
wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since head checksum not passed", pReader->pWal->cfg.vgId, wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since head checksum not passed", pReader->pWal->cfg.vgId,
ver); ver);
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED); TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED);
} }
@ -453,7 +453,7 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
if (pReader->capacity < cryptedBodyLen) { if (pReader->capacity < cryptedBodyLen) {
SWalCkHead *ptr = (SWalCkHead *)taosMemoryRealloc(pReader->pHead, sizeof(SWalCkHead) + cryptedBodyLen); SWalCkHead *ptr = (SWalCkHead *)taosMemoryRealloc(pReader->pHead, sizeof(SWalCkHead) + cryptedBodyLen);
if (ptr == NULL) { if (ptr == NULL) {
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY); TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
} }
@ -464,7 +464,7 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
if ((contLen = taosReadFile(pReader->pLogFile, pReader->pHead->head.body, cryptedBodyLen)) != cryptedBodyLen) { if ((contLen = taosReadFile(pReader->pLogFile, pReader->pHead->head.body, cryptedBodyLen)) != cryptedBodyLen) {
wError("vgId:%d, failed to read WAL record body, index:%" PRId64 ", from log file since %s", wError("vgId:%d, failed to read WAL record body, index:%" PRId64 ", from log file since %s",
pReader->pWal->cfg.vgId, ver, terrstr()); pReader->pWal->cfg.vgId, ver, terrstr());
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
if (contLen < 0) { if (contLen < 0) {
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
@ -477,14 +477,14 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
wError("vgId:%d, unexpected wal log, index:%" PRId64 ", read request index:%" PRId64, pReader->pWal->cfg.vgId, wError("vgId:%d, unexpected wal log, index:%" PRId64 ", read request index:%" PRId64, pReader->pWal->cfg.vgId,
pReader->pHead->head.version, ver); pReader->pHead->head.version, ver);
// pReader->curInvalid = 1; // pReader->curInvalid = 1;
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED); TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED);
} }
code = decryptBody(&pReader->pWal->cfg, pReader->pHead, plainBodyLen, __FUNCTION__); code = decryptBody(&pReader->pWal->cfg, pReader->pHead, plainBodyLen, __FUNCTION__);
if (code) { if (code) {
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -498,13 +498,13 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
wError("checksum written into log:%u, checksum calculated:%u", logCkSum, readCkSum); wError("checksum written into log:%u, checksum calculated:%u", logCkSum, readCkSum);
// pReader->curInvalid = 1; // pReader->curInvalid = 1;
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED); TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED);
} }
pReader->curVersion++; pReader->curVersion++;
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
} }
@ -523,13 +523,13 @@ int32_t decryptBody(SWalCfg *cfg, SWalCkHead *pHead, int32_t plainBodyLen, const
opts.source = pHead->head.body; opts.source = pHead->head.body;
opts.result = newBody; opts.result = newBody;
opts.unitLen = 16; opts.unitLen = 16;
strncpy(opts.key, cfg->encryptKey, 16); TAOS_UNUSED(strncpy((char *)opts.key, cfg->encryptKey, 16));
int32_t count = CBC_Decrypt(&opts); int32_t count = CBC_Decrypt(&opts);
// wDebug("CBC_Decrypt cryptedBodyLen:%d, plainBodyLen:%d, %s", count, plainBodyLen, func); // wDebug("CBC_Decrypt cryptedBodyLen:%d, plainBodyLen:%d, %s", count, plainBodyLen, func);
memcpy(pHead->head.body, newBody, plainBodyLen); TAOS_UNUSED(memcpy(pHead->head.body, newBody, plainBodyLen));
taosMemoryFree(newBody); taosMemoryFree(newBody);
} }
@ -538,10 +538,10 @@ int32_t decryptBody(SWalCfg *cfg, SWalCkHead *pHead, int32_t plainBodyLen, const
} }
void walReadReset(SWalReader *pReader) { void walReadReset(SWalReader *pReader) {
taosThreadMutexLock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexLock(&pReader->mutex));
taosCloseFile(&pReader->pIdxFile); TAOS_UNUSED(taosCloseFile(&pReader->pIdxFile));
taosCloseFile(&pReader->pLogFile); TAOS_UNUSED(taosCloseFile(&pReader->pLogFile));
pReader->curFileFirstVer = -1; pReader->curFileFirstVer = -1;
pReader->curVersion = -1; pReader->curVersion = -1;
taosThreadMutexUnlock(&pReader->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pReader->mutex));
} }

14
source/libs/wal/src/walRef.c
View File

@ -61,34 +61,34 @@ int32_t walSetRefVer(SWalRef *pRef, int64_t ver) {
SWal *pWal = pRef->pWal; SWal *pWal = pRef->pWal;
wDebug("vgId:%d, wal ref version %" PRId64 ", refId %" PRId64, pWal->cfg.vgId, ver, pRef->refId); wDebug("vgId:%d, wal ref version %" PRId64 ", refId %" PRId64, pWal->cfg.vgId, ver, pRef->refId);
if (pRef->refVer != ver) { if (pRef->refVer != ver) {
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
if (ver < pWal->vers.firstVer || ver > pWal->vers.lastVer) { if (ver < pWal->vers.firstVer || ver > pWal->vers.lastVer) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TSDB_CODE_WAL_INVALID_VER); TAOS_RETURN(TSDB_CODE_WAL_INVALID_VER);
} }
pRef->refVer = ver; pRef->refVer = ver;
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
} }
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
} }
void walRefFirstVer(SWal *pWal, SWalRef *pRef) { void walRefFirstVer(SWal *pWal, SWalRef *pRef) {
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
pRef->refVer = pWal->vers.firstVer; pRef->refVer = pWal->vers.firstVer;
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
wDebug("vgId:%d, wal ref version %" PRId64 " for first", pWal->cfg.vgId, pRef->refVer); wDebug("vgId:%d, wal ref version %" PRId64 " for first", pWal->cfg.vgId, pRef->refVer);
} }
void walRefLastVer(SWal *pWal, SWalRef *pRef) { void walRefLastVer(SWal *pWal, SWalRef *pRef) {
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
pRef->refVer = pWal->vers.lastVer; pRef->refVer = pWal->vers.lastVer;
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
wDebug("vgId:%d, wal ref version %" PRId64 " for last", pWal->cfg.vgId, pRef->refVer); wDebug("vgId:%d, wal ref version %" PRId64 " for last", pWal->cfg.vgId, pRef->refVer);
} }

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

@ -23,7 +23,7 @@
int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) { int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
wInfo("vgId:%d, restore from snapshot, version %" PRId64, pWal->cfg.vgId, ver); wInfo("vgId:%d, restore from snapshot, version %" PRId64, pWal->cfg.vgId, ver);
@ -34,14 +34,14 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
SWalRef *pRef = *(SWalRef **)pIter; SWalRef *pRef = *(SWalRef **)pIter;
if (pRef->refVer != -1 && pRef->refVer <= ver) { if (pRef->refVer != -1 && pRef->refVer <= ver) {
taosHashCancelIterate(pWal->pRefHash, pIter); taosHashCancelIterate(pWal->pRefHash, pIter);
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TSDB_CODE_FAILED); TAOS_RETURN(TSDB_CODE_FAILED);
} }
} }
taosCloseFile(&pWal->pLogFile); TAOS_UNUSED(taosCloseFile(&pWal->pLogFile));
taosCloseFile(&pWal->pIdxFile); TAOS_UNUSED(taosCloseFile(&pWal->pIdxFile));
if (pWal->vers.firstVer != -1) { if (pWal->vers.firstVer != -1) {
int32_t fileSetSize = taosArrayGetSize(pWal->fileInfoSet); int32_t fileSetSize = taosArrayGetSize(pWal->fileInfoSet);
@ -51,7 +51,7 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
walBuildLogName(pWal, pFileInfo->firstVer, fnameStr); walBuildLogName(pWal, pFileInfo->firstVer, fnameStr);
if (taosRemoveFile(fnameStr) < 0) { if (taosRemoveFile(fnameStr) < 0) {
wError("vgId:%d, restore from snapshot, cannot remove file %s since %s", pWal->cfg.vgId, fnameStr, terrstr()); wError("vgId:%d, restore from snapshot, cannot remove file %s since %s", pWal->cfg.vgId, fnameStr, terrstr());
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
@ -60,7 +60,7 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
walBuildIdxName(pWal, pFileInfo->firstVer, fnameStr); walBuildIdxName(pWal, pFileInfo->firstVer, fnameStr);
if (taosRemoveFile(fnameStr) < 0) { if (taosRemoveFile(fnameStr) < 0) {
wError("vgId:%d, cannot remove file %s since %s", pWal->cfg.vgId, fnameStr, terrstr()); wError("vgId:%d, cannot remove file %s since %s", pWal->cfg.vgId, fnameStr, terrstr());
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
@ -81,7 +81,7 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
pWal->vers.snapshotVer = ver; pWal->vers.snapshotVer = ver;
pWal->vers.verInSnapshotting = -1; pWal->vers.verInSnapshotting = -1;
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
} }
@ -146,7 +146,7 @@ static int64_t walChangeWrite(SWal *pWal, int64_t ver) {
walBuildLogName(pWal, fileFirstVer, fnameStr); walBuildLogName(pWal, fileFirstVer, fnameStr);
pLogTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND); pLogTFile = taosOpenFile(fnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pLogTFile == NULL) { if (pLogTFile == NULL) {
taosCloseFile(&pIdxTFile); TAOS_UNUSED(taosCloseFile(&pIdxTFile));
pWal->pLogFile = NULL; pWal->pLogFile = NULL;
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
@ -160,12 +160,12 @@ static int64_t walChangeWrite(SWal *pWal, int64_t ver) {
} }
int32_t walRollback(SWal *pWal, int64_t ver) { int32_t walRollback(SWal *pWal, int64_t ver) {
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
wInfo("vgId:%d, wal rollback for version %" PRId64, pWal->cfg.vgId, ver); wInfo("vgId:%d, wal rollback for version %" PRId64, pWal->cfg.vgId, ver);
int64_t code; int64_t code;
char fnameStr[WAL_FILE_LEN]; char fnameStr[WAL_FILE_LEN];
if (ver > pWal->vers.lastVer || ver <= pWal->vers.commitVer || ver <= pWal->vers.snapshotVer) { if (ver > pWal->vers.lastVer || ver <= pWal->vers.commitVer || ver <= pWal->vers.snapshotVer) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TSDB_CODE_WAL_INVALID_VER); TAOS_RETURN(TSDB_CODE_WAL_INVALID_VER);
} }
@ -175,7 +175,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
// change current files // change current files
code = walChangeWrite(pWal, ver); code = walChangeWrite(pWal, ver);
if (code < 0) { if (code < 0) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -187,50 +187,50 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
walBuildLogName(pWal, pInfo->firstVer, fnameStr); walBuildLogName(pWal, pInfo->firstVer, fnameStr);
wDebug("vgId:%d, wal remove file %s for rollback", pWal->cfg.vgId, fnameStr); wDebug("vgId:%d, wal remove file %s for rollback", pWal->cfg.vgId, fnameStr);
taosRemoveFile(fnameStr); TAOS_UNUSED(taosRemoveFile(fnameStr));
walBuildIdxName(pWal, pInfo->firstVer, fnameStr); walBuildIdxName(pWal, pInfo->firstVer, fnameStr);
wDebug("vgId:%d, wal remove file %s for rollback", pWal->cfg.vgId, fnameStr); wDebug("vgId:%d, wal remove file %s for rollback", pWal->cfg.vgId, fnameStr);
taosRemoveFile(fnameStr); TAOS_UNUSED(taosRemoveFile(fnameStr));
} }
} }
walBuildIdxName(pWal, walGetCurFileFirstVer(pWal), fnameStr); walBuildIdxName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
taosCloseFile(&pWal->pIdxFile); TAOS_UNUSED(taosCloseFile(&pWal->pIdxFile));
TdFilePtr pIdxFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND); TdFilePtr pIdxFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
if (pIdxFile == NULL) { if (pIdxFile == NULL) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
int64_t idxOff = walGetVerIdxOffset(pWal, ver); int64_t idxOff = walGetVerIdxOffset(pWal, ver);
code = taosLSeekFile(pIdxFile, idxOff, SEEK_SET); code = taosLSeekFile(pIdxFile, idxOff, SEEK_SET);
if (code < 0) { if (code < 0) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
// read idx file and get log file pos // read idx file and get log file pos
SWalIdxEntry entry; SWalIdxEntry entry;
if (taosReadFile(pIdxFile, &entry, sizeof(SWalIdxEntry)) != sizeof(SWalIdxEntry)) { if (taosReadFile(pIdxFile, &entry, sizeof(SWalIdxEntry)) != sizeof(SWalIdxEntry)) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr); walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
taosCloseFile(&pWal->pLogFile); TAOS_UNUSED(taosCloseFile(&pWal->pLogFile));
TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND); TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
wDebug("vgId:%d, wal truncate file %s", pWal->cfg.vgId, fnameStr); wDebug("vgId:%d, wal truncate file %s", pWal->cfg.vgId, fnameStr);
if (pLogFile == NULL) { if (pLogFile == NULL) {
// TODO // TODO
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
code = taosLSeekFile(pLogFile, entry.offset, SEEK_SET); code = taosLSeekFile(pLogFile, entry.offset, SEEK_SET);
if (code < 0) { if (code < 0) {
// TODO // TODO
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
@ -238,19 +238,19 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
SWalCkHead head; SWalCkHead head;
int64_t size = taosReadFile(pLogFile, &head, sizeof(SWalCkHead)); int64_t size = taosReadFile(pLogFile, &head, sizeof(SWalCkHead));
if (size != sizeof(SWalCkHead)) { if (size != sizeof(SWalCkHead)) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
code = walValidHeadCksum(&head); code = walValidHeadCksum(&head);
if (code != 0) { if (code != 0) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED); TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED);
} }
if (head.head.version != ver) { if (head.head.version != ver) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED); TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED);
} }
@ -258,13 +258,13 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
// truncate old files // truncate old files
code = taosFtruncateFile(pLogFile, entry.offset); code = taosFtruncateFile(pLogFile, entry.offset);
if (code < 0) { if (code < 0) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
code = taosFtruncateFile(pIdxFile, idxOff); code = taosFtruncateFile(pIdxFile, idxOff);
if (code < 0) { if (code < 0) {
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno)); TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
} }
@ -272,19 +272,19 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->lastVer = ver - 1; ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->lastVer = ver - 1;
((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize = entry.offset; ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize = entry.offset;
taosCloseFile(&pIdxFile); TAOS_UNUSED(taosCloseFile(&pIdxFile));
taosCloseFile(&pLogFile); TAOS_UNUSED(taosCloseFile(&pLogFile));
code = walSaveMeta(pWal); code = walSaveMeta(pWal);
if (code < 0) { if (code < 0) {
wError("vgId:%d, failed to save meta since %s", pWal->cfg.vgId, terrstr()); wError("vgId:%d, failed to save meta since %s", pWal->cfg.vgId, terrstr());
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(code); TAOS_RETURN(code);
} }
// unlock // unlock
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
} }
@ -371,7 +371,7 @@ static FORCE_INLINE int32_t walCheckAndRoll(SWal *pWal) {
int32_t walBeginSnapshot(SWal *pWal, int64_t ver, int64_t logRetention) { int32_t walBeginSnapshot(SWal *pWal, int64_t ver, int64_t logRetention) {
int32_t code = 0; int32_t code = 0;
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
ASSERT(logRetention >= 0); ASSERT(logRetention >= 0);
pWal->vers.verInSnapshotting = ver; pWal->vers.verInSnapshotting = ver;
pWal->vers.logRetention = logRetention; pWal->vers.logRetention = logRetention;
@ -388,7 +388,7 @@ int32_t walBeginSnapshot(SWal *pWal, int64_t ver, int64_t logRetention) {
} }
_exit: _exit:
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -396,7 +396,7 @@ _exit:
int32_t walEndSnapshot(SWal *pWal) { int32_t walEndSnapshot(SWal *pWal) {
int32_t code = 0, lino = 0; int32_t code = 0, lino = 0;
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
int64_t ver = pWal->vers.verInSnapshotting; int64_t ver = pWal->vers.verInSnapshotting;
wDebug("vgId:%d, wal end snapshot for version %" PRId64 ", log retention %" PRId64 " first ver %" PRId64 wDebug("vgId:%d, wal end snapshot for version %" PRId64 ", log retention %" PRId64 " first ver %" PRId64
@ -456,7 +456,7 @@ int32_t walEndSnapshot(SWal *pWal) {
} }
for (SWalFileInfo *iter = pWal->fileInfoSet->pData; iter <= pUntil; iter++) { for (SWalFileInfo *iter = pWal->fileInfoSet->pData; iter <= pUntil; iter++) {
deleteCnt++; deleteCnt++;
taosArrayPush(pWal->toDeleteFiles, iter); TAOS_UNUSED(taosArrayPush(pWal->toDeleteFiles, iter));
} }
// make new array, remove files // make new array, remove files
@ -589,8 +589,8 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit); TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
} }
memset(newBody, 0, cyptedBodyLen); TAOS_UNUSED(memset(newBody, 0, cyptedBodyLen));
memcpy(newBody, body, plainBodyLen); TAOS_UNUSED(memcpy(newBody, body, plainBodyLen));
newBodyEncrypted = taosMemoryMalloc(cyptedBodyLen); newBodyEncrypted = taosMemoryMalloc(cyptedBodyLen);
if (newBodyEncrypted == NULL) { if (newBodyEncrypted == NULL) {
@ -607,7 +607,7 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
opts.source = newBody; opts.source = newBody;
opts.result = newBodyEncrypted; opts.result = newBodyEncrypted;
opts.unitLen = 16; opts.unitLen = 16;
strncpy(opts.key, pWal->cfg.encryptKey, ENCRYPT_KEY_LEN); TAOS_UNUSED(strncpy((char *)opts.key, pWal->cfg.encryptKey, ENCRYPT_KEY_LEN));
int32_t count = CBC_Encrypt(&opts); int32_t count = CBC_Encrypt(&opts);
@ -698,7 +698,7 @@ int32_t walAppendLog(SWal *pWal, int64_t index, tmsg_t msgType, SWalSyncInfo syn
int32_t bodyLen) { int32_t bodyLen) {
int32_t code = 0, lino = 0; int32_t code = 0, lino = 0;
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
if (index != pWal->vers.lastVer + 1) { if (index != pWal->vers.lastVer + 1) {
TAOS_CHECK_GOTO(TSDB_CODE_WAL_INVALID_VER, &lino, _exit); TAOS_CHECK_GOTO(TSDB_CODE_WAL_INVALID_VER, &lino, _exit);
@ -717,7 +717,7 @@ _exit:
wError("%s failed at line %d since %s", __func__, lino, tstrerror(code)); wError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
} }
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
return code; return code;
} }
@ -728,7 +728,7 @@ int32_t walFsync(SWal *pWal, bool forceFsync) {
return code; return code;
} }
taosThreadMutexLock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexLock(&pWal->mutex));
if (forceFsync || (pWal->cfg.level == TAOS_WAL_FSYNC && pWal->cfg.fsyncPeriod == 0)) { if (forceFsync || (pWal->cfg.level == TAOS_WAL_FSYNC && pWal->cfg.fsyncPeriod == 0)) {
wTrace("vgId:%d, fileId:%" PRId64 ".log, do fsync", pWal->cfg.vgId, walGetCurFileFirstVer(pWal)); wTrace("vgId:%d, fileId:%" PRId64 ".log, do fsync", pWal->cfg.vgId, walGetCurFileFirstVer(pWal));
if (taosFsyncFile(pWal->pLogFile) < 0) { if (taosFsyncFile(pWal->pLogFile) < 0) {
@ -737,7 +737,7 @@ int32_t walFsync(SWal *pWal, bool forceFsync) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
} }
} }
taosThreadMutexUnlock(&pWal->mutex); TAOS_UNUSED(taosThreadMutexUnlock(&pWal->mutex));
return code; return code;
} }

76
source/util/src/tlrucache.c
View File

@ -333,13 +333,13 @@ static void taosLRUCacheShardEvictLRU(SLRUCacheShard *shard, size_t charge, SArr
static void taosLRUCacheShardSetCapacity(SLRUCacheShard *shard, size_t capacity) { static void taosLRUCacheShardSetCapacity(SLRUCacheShard *shard, size_t capacity) {
SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES); SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES);
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
shard->capacity = capacity; shard->capacity = capacity;
shard->highPriPoolCapacity = capacity * shard->highPriPoolRatio; shard->highPriPoolCapacity = capacity * shard->highPriPoolRatio;
taosLRUCacheShardEvictLRU(shard, 0, lastReferenceList); taosLRUCacheShardEvictLRU(shard, 0, lastReferenceList);
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) { for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) {
SLRUEntry *entry = taosArrayGetP(lastReferenceList, i); SLRUEntry *entry = taosArrayGetP(lastReferenceList, i);
@ -352,9 +352,9 @@ static int taosLRUCacheShardInit(SLRUCacheShard *shard, size_t capacity, bool st
int maxUpperHashBits) { int maxUpperHashBits) {
TAOS_CHECK_RETURN(taosLRUEntryTableInit(&shard->table, maxUpperHashBits)); TAOS_CHECK_RETURN(taosLRUEntryTableInit(&shard->table, maxUpperHashBits));
taosThreadMutexInit(&shard->mutex, NULL); (void)taosThreadMutexInit(&shard->mutex, NULL);
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
shard->capacity = 0; shard->capacity = 0;
shard->highPriPoolUsage = 0; shard->highPriPoolUsage = 0;
shard->strictCapacity = strict; shard->strictCapacity = strict;
@ -367,7 +367,7 @@ static int taosLRUCacheShardInit(SLRUCacheShard *shard, size_t capacity, bool st
shard->lru.next = &shard->lru; shard->lru.next = &shard->lru;
shard->lru.prev = &shard->lru; shard->lru.prev = &shard->lru;
shard->lruLowPri = &shard->lru; shard->lruLowPri = &shard->lru;
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
taosLRUCacheShardSetCapacity(shard, capacity); taosLRUCacheShardSetCapacity(shard, capacity);
@ -375,7 +375,7 @@ static int taosLRUCacheShardInit(SLRUCacheShard *shard, size_t capacity, bool st
} }
static void taosLRUCacheShardCleanup(SLRUCacheShard *shard) { static void taosLRUCacheShardCleanup(SLRUCacheShard *shard) {
taosThreadMutexDestroy(&shard->mutex); (void)taosThreadMutexDestroy(&shard->mutex);
taosLRUEntryTableCleanup(&shard->table); taosLRUEntryTableCleanup(&shard->table);
} }
@ -385,7 +385,7 @@ static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry *
LRUStatus status = TAOS_LRU_STATUS_OK; LRUStatus status = TAOS_LRU_STATUS_OK;
SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES); SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES);
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
taosLRUCacheShardEvictLRU(shard, e->totalCharge, lastReferenceList); taosLRUCacheShardEvictLRU(shard, e->totalCharge, lastReferenceList);
@ -429,7 +429,7 @@ static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry *
} }
} }
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) { for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) {
SLRUEntry *entry = taosArrayGetP(lastReferenceList, i); SLRUEntry *entry = taosArrayGetP(lastReferenceList, i);
@ -470,7 +470,7 @@ static LRUStatus taosLRUCacheShardInsert(SLRUCacheShard *shard, const void *key,
static LRUHandle *taosLRUCacheShardLookup(SLRUCacheShard *shard, const void *key, size_t keyLen, uint32_t hash) { static LRUHandle *taosLRUCacheShardLookup(SLRUCacheShard *shard, const void *key, size_t keyLen, uint32_t hash) {
SLRUEntry *e = NULL; SLRUEntry *e = NULL;
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
e = taosLRUEntryTableLookup(&shard->table, key, keyLen, hash); e = taosLRUEntryTableLookup(&shard->table, key, keyLen, hash);
if (e != NULL) { if (e != NULL) {
ASSERT(TAOS_LRU_ENTRY_IN_CACHE(e)); ASSERT(TAOS_LRU_ENTRY_IN_CACHE(e));
@ -481,14 +481,14 @@ static LRUHandle *taosLRUCacheShardLookup(SLRUCacheShard *shard, const void *key
TAOS_LRU_ENTRY_SET_HIT(e); TAOS_LRU_ENTRY_SET_HIT(e);
} }
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
return (LRUHandle *)e; return (LRUHandle *)e;
} }
static void taosLRUCacheShardErase(SLRUCacheShard *shard, const void *key, size_t keyLen, uint32_t hash) { static void taosLRUCacheShardErase(SLRUCacheShard *shard, const void *key, size_t keyLen, uint32_t hash) {
bool lastReference = false; bool lastReference = false;
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
SLRUEntry *e = taosLRUEntryTableRemove(&shard->table, key, keyLen, hash); SLRUEntry *e = taosLRUEntryTableRemove(&shard->table, key, keyLen, hash);
if (e != NULL) { if (e != NULL) {
@ -503,7 +503,7 @@ static void taosLRUCacheShardErase(SLRUCacheShard *shard, const void *key, size_
} }
} }
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
if (lastReference) { if (lastReference) {
taosLRUEntryFree(e); taosLRUEntryFree(e);
@ -513,11 +513,11 @@ static void taosLRUCacheShardErase(SLRUCacheShard *shard, const void *key, size_
static int taosLRUCacheShardApply(SLRUCacheShard *shard, _taos_lru_functor_t functor, void *ud) { static int taosLRUCacheShardApply(SLRUCacheShard *shard, _taos_lru_functor_t functor, void *ud) {
int ret; int ret;
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
ret = taosLRUEntryTableApplyF(&shard->table, functor, ud); ret = taosLRUEntryTableApplyF(&shard->table, functor, ud);
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
return ret; return ret;
} }
@ -525,7 +525,7 @@ static int taosLRUCacheShardApply(SLRUCacheShard *shard, _taos_lru_functor_t fun
static void taosLRUCacheShardEraseUnrefEntries(SLRUCacheShard *shard) { static void taosLRUCacheShardEraseUnrefEntries(SLRUCacheShard *shard) {
SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES); SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES);
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
while (shard->lru.next != &shard->lru) { while (shard->lru.next != &shard->lru) {
SLRUEntry *old = shard->lru.next; SLRUEntry *old = shard->lru.next;
@ -539,7 +539,7 @@ static void taosLRUCacheShardEraseUnrefEntries(SLRUCacheShard *shard) {
taosArrayPush(lastReferenceList, &old); taosArrayPush(lastReferenceList, &old);
} }
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) { for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) {
SLRUEntry *entry = taosArrayGetP(lastReferenceList, i); SLRUEntry *entry = taosArrayGetP(lastReferenceList, i);
@ -552,12 +552,12 @@ static void taosLRUCacheShardEraseUnrefEntries(SLRUCacheShard *shard) {
static bool taosLRUCacheShardRef(SLRUCacheShard *shard, LRUHandle *handle) { static bool taosLRUCacheShardRef(SLRUCacheShard *shard, LRUHandle *handle) {
SLRUEntry *e = (SLRUEntry *)handle; SLRUEntry *e = (SLRUEntry *)handle;
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
ASSERT(TAOS_LRU_ENTRY_HAS_REFS(e)); ASSERT(TAOS_LRU_ENTRY_HAS_REFS(e));
TAOS_LRU_ENTRY_REF(e); TAOS_LRU_ENTRY_REF(e);
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
return true; return true;
} }
@ -570,7 +570,7 @@ static bool taosLRUCacheShardRelease(SLRUCacheShard *shard, LRUHandle *handle, b
SLRUEntry *e = (SLRUEntry *)handle; SLRUEntry *e = (SLRUEntry *)handle;
bool lastReference = false; bool lastReference = false;
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
lastReference = taosLRUEntryUnref(e); lastReference = taosLRUEntryUnref(e);
if (lastReference && TAOS_LRU_ENTRY_IN_CACHE(e)) { if (lastReference && TAOS_LRU_ENTRY_IN_CACHE(e)) {
@ -591,7 +591,7 @@ static bool taosLRUCacheShardRelease(SLRUCacheShard *shard, LRUHandle *handle, b
shard->usage -= e->totalCharge; shard->usage -= e->totalCharge;
} }
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
if (lastReference) { if (lastReference) {
taosLRUEntryFree(e); taosLRUEntryFree(e);
@ -603,9 +603,9 @@ static bool taosLRUCacheShardRelease(SLRUCacheShard *shard, LRUHandle *handle, b
static size_t taosLRUCacheShardGetUsage(SLRUCacheShard *shard) { static size_t taosLRUCacheShardGetUsage(SLRUCacheShard *shard) {
size_t usage = 0; size_t usage = 0;
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
usage = shard->usage; usage = shard->usage;
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
return usage; return usage;
} }
@ -613,9 +613,9 @@ static size_t taosLRUCacheShardGetUsage(SLRUCacheShard *shard) {
static int32_t taosLRUCacheShardGetElems(SLRUCacheShard *shard) { static int32_t taosLRUCacheShardGetElems(SLRUCacheShard *shard) {
int32_t elems = 0; int32_t elems = 0;
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
elems = shard->table.elems; elems = shard->table.elems;
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
return elems; return elems;
} }
@ -623,22 +623,22 @@ static int32_t taosLRUCacheShardGetElems(SLRUCacheShard *shard) {
static size_t taosLRUCacheShardGetPinnedUsage(SLRUCacheShard *shard) { static size_t taosLRUCacheShardGetPinnedUsage(SLRUCacheShard *shard) {
size_t usage = 0; size_t usage = 0;
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
ASSERT(shard->usage >= shard->lruUsage); ASSERT(shard->usage >= shard->lruUsage);
usage = shard->usage - shard->lruUsage; usage = shard->usage - shard->lruUsage;
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
return usage; return usage;
} }
static void taosLRUCacheShardSetStrictCapacity(SLRUCacheShard *shard, bool strict) { static void taosLRUCacheShardSetStrictCapacity(SLRUCacheShard *shard, bool strict) {
taosThreadMutexLock(&shard->mutex); (void)taosThreadMutexLock(&shard->mutex);
shard->strictCapacity = strict; shard->strictCapacity = strict;
taosThreadMutexUnlock(&shard->mutex); (void)taosThreadMutexUnlock(&shard->mutex);
} }
struct SShardedCache { struct SShardedCache {
@ -706,7 +706,7 @@ SLRUCache *taosLRUCacheInit(size_t capacity, int numShardBits, double highPriPoo
cache->shardedCache.capacity = capacity; cache->shardedCache.capacity = capacity;
cache->shardedCache.lastId = 1; cache->shardedCache.lastId = 1;
taosThreadMutexInit(&cache->shardedCache.capacityMutex, NULL); (void)taosThreadMutexInit(&cache->shardedCache.capacityMutex, NULL);
return cache; return cache;
} }
@ -723,7 +723,7 @@ void taosLRUCacheCleanup(SLRUCache *cache) {
cache->shards = 0; cache->shards = 0;
} }
taosThreadMutexDestroy(&cache->shardedCache.capacityMutex); (void)taosThreadMutexDestroy(&cache->shardedCache.capacityMutex);
taosMemoryFree(cache); taosMemoryFree(cache);
} }
@ -826,7 +826,7 @@ void taosLRUCacheSetCapacity(SLRUCache *cache, size_t capacity) {
uint32_t numShards = cache->numShards; uint32_t numShards = cache->numShards;
size_t perShard = (capacity + (numShards - 1)) / numShards; size_t perShard = (capacity + (numShards - 1)) / numShards;
taosThreadMutexLock(&cache->shardedCache.capacityMutex); (void)taosThreadMutexLock(&cache->shardedCache.capacityMutex);
for (int i = 0; i < numShards; ++i) { for (int i = 0; i < numShards; ++i) {
taosLRUCacheShardSetCapacity(&cache->shards[i], perShard); taosLRUCacheShardSetCapacity(&cache->shards[i], perShard);
@ -834,17 +834,17 @@ void taosLRUCacheSetCapacity(SLRUCache *cache, size_t capacity) {
cache->shardedCache.capacity = capacity; cache->shardedCache.capacity = capacity;
taosThreadMutexUnlock(&cache->shardedCache.capacityMutex); (void)taosThreadMutexUnlock(&cache->shardedCache.capacityMutex);
} }
size_t taosLRUCacheGetCapacity(SLRUCache *cache) { size_t taosLRUCacheGetCapacity(SLRUCache *cache) {
size_t capacity = 0; size_t capacity = 0;
taosThreadMutexLock(&cache->shardedCache.capacityMutex); (void)taosThreadMutexLock(&cache->shardedCache.capacityMutex);
capacity = cache->shardedCache.capacity; capacity = cache->shardedCache.capacity;
taosThreadMutexUnlock(&cache->shardedCache.capacityMutex); (void)taosThreadMutexUnlock(&cache->shardedCache.capacityMutex);
return capacity; return capacity;
} }
@ -852,7 +852,7 @@ size_t taosLRUCacheGetCapacity(SLRUCache *cache) {
void taosLRUCacheSetStrictCapacity(SLRUCache *cache, bool strict) { void taosLRUCacheSetStrictCapacity(SLRUCache *cache, bool strict) {
uint32_t numShards = cache->numShards; uint32_t numShards = cache->numShards;
taosThreadMutexLock(&cache->shardedCache.capacityMutex); (void)taosThreadMutexLock(&cache->shardedCache.capacityMutex);
for (int i = 0; i < numShards; ++i) { for (int i = 0; i < numShards; ++i) {
taosLRUCacheShardSetStrictCapacity(&cache->shards[i], strict); taosLRUCacheShardSetStrictCapacity(&cache->shards[i], strict);
@ -860,17 +860,17 @@ void taosLRUCacheSetStrictCapacity(SLRUCache *cache, bool strict) {
cache->shardedCache.strictCapacity = strict; cache->shardedCache.strictCapacity = strict;
taosThreadMutexUnlock(&cache->shardedCache.capacityMutex); (void)taosThreadMutexUnlock(&cache->shardedCache.capacityMutex);
} }
bool taosLRUCacheIsStrictCapacity(SLRUCache *cache) { bool taosLRUCacheIsStrictCapacity(SLRUCache *cache) {
bool strict = false; bool strict = false;
taosThreadMutexLock(&cache->shardedCache.capacityMutex); (void)taosThreadMutexLock(&cache->shardedCache.capacityMutex);
strict = cache->shardedCache.strictCapacity; strict = cache->shardedCache.strictCapacity;
taosThreadMutexUnlock(&cache->shardedCache.capacityMutex); (void)taosThreadMutexUnlock(&cache->shardedCache.capacityMutex);
return strict; return strict;
} }