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

Merge pull request #27962 from taosdata/enh/removeVoidOfTransport 

remove void
This commit is contained in:
Hongze Cheng 2024-09-24 17:12:03 +08:00 committed by GitHub
parent 8037f835b9 44cac014d6
commit c824a70b83
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 283 additions and 289 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/dnode/vnode/src/tq/tqStreamTaskSnap.c
View File

@ -137,7 +137,7 @@ NextTbl:
memcpy(pVal, tVal, tLen);
vLen = tLen;
}
(void)tdbTbcMoveToNext(pReader->pCur);
TAOS_UNUSED(tdbTbcMoveToNext(pReader->pCur));
break;
}
if (except == 1) {
@ -147,7 +147,7 @@ NextTbl:
pReader->pos += 1;
pPair = taosArrayGet(pReader->tdbTbList, pReader->pos);
code = tdbTbcOpen(pPair->tbl, &pReader->pCur, NULL);
(void)tdbTbcMoveToFirst(pReader->pCur);
TAOS_UNUSED(tdbTbcMoveToFirst(pReader->pCur));
goto NextTbl;
}
@ -210,7 +210,7 @@ int32_t streamTaskSnapWriterClose(SStreamTaskWriter* pWriter, int8_t rollback, i
streamMetaWLock(pTq->pStreamMeta);
tqDebug("vgId:%d, vnode stream-task snapshot writer closed", TD_VID(pTq->pVnode));
if (rollback) {
tdbAbort(pTq->pStreamMeta->db, pTq->pStreamMeta->txn);
TAOS_UNUSED(tdbAbort(pTq->pStreamMeta->db, pTq->pStreamMeta->txn));
} else {
code = tdbCommit(pTq->pStreamMeta->db, pTq->pStreamMeta->txn);
if (code) goto _err;

206
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -427,7 +427,7 @@ void cleanDir(const char* pPath, const char* id) {
if (taosIsDir(pPath)) {
taosRemoveDir(pPath);
(void)taosMkDir(pPath);
TAOS_UNUSED(taosMkDir(pPath));
stInfo("%s clear dir:%s, succ", id, pPath);
}
}
@ -533,7 +533,7 @@ int32_t rebuildFromRemoteChkp_s3(const char* key, char* chkpPath, int64_t chkpId
_EXIT:
if (code != 0) {
if (rename) {
(void)taosRenameFile(defaultTmp, defaultPath);
TAOS_UNUSED(taosRenameFile(defaultTmp, defaultPath));
}
}
@ -653,13 +653,13 @@ int32_t backendFileCopyFilesImpl(const char* src, const char* dst) {
taosMemoryFreeClear(srcName);
taosMemoryFreeClear(dstName);
(void)taosCloseDir(&pDir);
TAOS_UNUSED(taosCloseDir(&pDir));
return code;
_ERROR:
taosMemoryFreeClear(srcName);
taosMemoryFreeClear(dstName);
(void)taosCloseDir(&pDir);
TAOS_UNUSED(taosCloseDir(&pDir));
return code;
}
@ -833,9 +833,8 @@ int32_t streamBackendInit(const char* streamPath, int64_t chkpId, int32_t vgId,
pHandle->list = tdListNew(sizeof(SCfComparator));
TSDB_CHECK_NULL(pHandle->list, code, lino, _EXIT, terrno);
(void)taosThreadMutexInit(&pHandle->mutex, NULL);
(void)taosThreadMutexInit(&pHandle->cfMutex, NULL);
TAOS_UNUSED(taosThreadMutexInit(&pHandle->mutex, NULL));
TAOS_UNUSED(taosThreadMutexInit(&pHandle->cfMutex, NULL));
pHandle->cfInst = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
TSDB_CHECK_NULL(pHandle->cfInst, code, lino, _EXIT, terrno);
@ -948,11 +947,11 @@ void streamBackendCleanup(void* arg) {
void streamBackendHandleCleanup(void* arg) {
SBackendCfWrapper* wrapper = arg;
bool remove = wrapper->remove;
(void)taosThreadRwlockWrlock(&wrapper->rwLock);
TAOS_UNUSED(taosThreadRwlockWrlock(&wrapper->rwLock));
stDebug("start to do-close backendwrapper %p, %s", wrapper, wrapper->idstr);
if (wrapper->rocksdb == NULL) {
(void)taosThreadRwlockUnlock(&wrapper->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&wrapper->rwLock));
return;
}
@ -1003,9 +1002,9 @@ void streamBackendHandleCleanup(void* arg) {
wrapper->readOpts = NULL;
taosMemoryFreeClear(wrapper->cfOpts);
taosMemoryFreeClear(wrapper->param);
(void)taosThreadRwlockUnlock(&wrapper->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&wrapper->rwLock));
(void)taosThreadRwlockDestroy(&wrapper->rwLock);
TAOS_UNUSED(taosThreadRwlockDestroy(&wrapper->rwLock));
wrapper->rocksdb = NULL;
// taosReleaseRef(streamBackendId, wrapper->backendId);
@ -1099,7 +1098,7 @@ int32_t chkpMayDelObsolete(void* arg, int64_t chkpId, char* path) {
int32_t code = 0;
STaskDbWrapper* pBackend = arg;
SArray * chkpDel = NULL, *chkpDup = NULL;
(void)taosThreadRwlockWrlock(&pBackend->chkpDirLock);
TAOS_UNUSED(taosThreadRwlockWrlock(&pBackend->chkpDirLock));
if (taosArrayPush(pBackend->chkpSaved, &chkpId) == NULL) {
TAOS_CHECK_GOTO(terrno, NULL, _exception);
@ -1152,7 +1151,7 @@ int32_t chkpMayDelObsolete(void* arg, int64_t chkpId, char* path) {
taosArrayDestroy(pBackend->chkpSaved);
pBackend->chkpSaved = chkpDup;
(void)taosThreadRwlockUnlock(&pBackend->chkpDirLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&pBackend->chkpDirLock));
for (int i = 0; i < taosArrayGetSize(chkpDel); i++) {
int64_t id = *(int64_t*)taosArrayGet(chkpDel, i);
@ -1169,7 +1168,7 @@ int32_t chkpMayDelObsolete(void* arg, int64_t chkpId, char* path) {
_exception:
taosArrayDestroy(chkpDup);
taosArrayDestroy(chkpDel);
(void)taosThreadRwlockUnlock(&pBackend->chkpDirLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&pBackend->chkpDirLock));
return code;
}
@ -1315,12 +1314,12 @@ int32_t taskDbLoadChkpInfo(STaskDbWrapper* pBackend) {
taosArraySort(pBackend->chkpSaved, chkpIdComp);
taosMemoryFree(pChkpDir);
(void)taosCloseDir(&pDir);
TAOS_UNUSED(taosCloseDir(&pDir));
return 0;
_exception:
taosMemoryFree(pChkpDir);
(void)taosCloseDir(&pDir);
TAOS_UNUSED(taosCloseDir(&pDir));
return code;
}
int32_t chkpGetAllDbCfHandle2(STaskDbWrapper* pBackend, rocksdb_column_family_handle_t*** ppHandle) {
@ -1561,7 +1560,7 @@ void* taskAcquireDb(int64_t refId) {
}
void taskReleaseDb(int64_t refId) {
// release
(void)taosReleaseRef(taskDbWrapperId, refId);
TAOS_UNUSED(taosReleaseRef(taskDbWrapperId, refId));
}
int64_t taskGetDBRef(void* arg) {
@ -1622,7 +1621,7 @@ int32_t chkpLoadExtraInfo(char* pChkpIdDir, int64_t* chkpId, int64_t* processId)
code = 0;
_EXIT:
taosMemoryFree(pDst);
(void)taosCloseFile(&pFile);
TAOS_UNUSED(taosCloseFile(&pFile));
return code;
}
int32_t chkpAddExtraInfo(char* pChkpIdDir, int64_t chkpId, int64_t processId) {
@ -1677,7 +1676,7 @@ int32_t chkpAddExtraInfo(char* pChkpIdDir, int64_t chkpId, int64_t processId) {
code = 0;
_EXIT:
(void)taosCloseFile(&pFile);
TAOS_UNUSED(taosCloseFile(&pFile));
taosMemoryFree(pDst);
return code;
}
@ -1735,7 +1734,7 @@ int32_t taskDbDoCheckpoint(void* arg, int64_t chkpId, int64_t processId) {
goto _EXIT;
}
(void)atomic_store_64(&pTaskDb->dataWritten, 0);
TAOS_UNUSED(atomic_store_64(&pTaskDb->dataWritten, 0));
pTaskDb->chkpId = chkpId;
_EXIT:
@ -1743,13 +1742,13 @@ _EXIT:
// clear checkpoint dir if failed
if (code != 0 && pChkpDir != NULL) {
if (taosDirExist(pChkpIdDir)) {
(void)taosRemoveDir(pChkpIdDir);
TAOS_UNUSED(taosRemoveDir(pChkpIdDir));
}
}
taosMemoryFree(pChkpIdDir);
taosMemoryFree(pChkpDir);
(void)taosReleaseRef(taskDbWrapperId, refId);
TAOS_UNUSED(taosReleaseRef(taskDbWrapperId, refId));
taosMemoryFree(ppCf);
return code;
}
@ -1822,7 +1821,7 @@ int defaultKeyComp(void* state, const char* aBuf, size_t aLen, const char* bBuf,
}
int streamStateValueIsStale(char* v) {
int64_t ts = 0;
(void)taosDecodeFixedI64(v, &ts);
TAOS_UNUSED(taosDecodeFixedI64(v, &ts));
return (ts != 0 && ts < taosGetTimestampMs()) ? 1 : 0;
}
int iterValueIsStale(rocksdb_iterator_t* iter) {
@ -1865,8 +1864,8 @@ int stateKeyDBComp(void* state, const char* aBuf, size_t aLen, const char* bBuf,
p1 = taosDecodeFixedI64(p1, &key1.key.ts);
p2 = taosDecodeFixedI64(p2, &key2.key.ts);
(void)taosDecodeFixedI64(p1, &key1.opNum);
(void)taosDecodeFixedI64(p2, &key2.opNum);
TAOS_UNUSED(taosDecodeFixedI64(p1, &key1.opNum));
TAOS_UNUSED(taosDecodeFixedI64(p2, &key2.opNum));
return stateKeyCmpr(&key1, sizeof(key1), &key2, sizeof(key2));
}
@ -2062,8 +2061,8 @@ int parKeyDBComp(void* state, const char* aBuf, size_t aLen, const char* bBuf, s
char* p1 = (char*)aBuf;
char* p2 = (char*)bBuf;
(void)taosDecodeFixedI64(p1, &w1);
(void)taosDecodeFixedI64(p2, &w2);
TAOS_UNUSED(taosDecodeFixedI64(p1, &w1));
TAOS_UNUSED(taosDecodeFixedI64(p2, &w2));
if (w1 == w2) {
return 0;
} else {
@ -2107,7 +2106,7 @@ int32_t valueIsStale(void* k, int64_t ts) {
}
void destroyCompare(void* arg) {
(void)arg;
TAOS_UNUSED(arg);
return;
}
@ -2251,31 +2250,28 @@ _EXCEPT:
}
const char* compareDefaultName(void* arg) {
(void)arg;
TAOS_UNUSED(arg);
return ginitDict[0].key;
}
const char* compareStateName(void* arg) {
(void)arg;
TAOS_UNUSED(arg);
return ginitDict[1].key;
}
const char* compareWinKeyName(void* arg) {
(void)arg;
return ginitDict[2].key;
}
const char* compareWinKeyName(void* arg) { return ginitDict[2].key; }
const char* compareSessionKeyName(void* arg) {
(void)arg;
TAOS_UNUSED(arg);
return ginitDict[3].key;
}
const char* compareFuncKeyName(void* arg) {
(void)arg;
TAOS_UNUSED(arg);
return ginitDict[4].key;
}
const char* compareParKeyName(void* arg) {
(void)arg;
TAOS_UNUSED(arg);
return ginitDict[5].key;
}
const char* comparePartagKeyName(void* arg) {
(void)arg;
TAOS_UNUSED(arg);
return ginitDict[6].key;
}
@ -2303,7 +2299,7 @@ const char* compactFilteFactoryNameFunc(void* arg) {
return "stream_compact_factory_filter_func";
}
void destroyCompactFilte(void* arg) { (void)arg; }
void destroyCompactFilte(void* arg) { TAOS_UNUSED(arg); }
unsigned char compactFilte(void* arg, int level, const char* key, size_t klen, const char* val, size_t vlen,
char** newval, size_t* newvlen, unsigned char* value_changed) {
return streamStateValueIsStale((char*)val) ? 1 : 0;
@ -2416,7 +2412,7 @@ void taskDbRemoveRef(void* pTaskDb) {
}
STaskDbWrapper* pBackend = pTaskDb;
(void)taosReleaseRef(taskDbWrapperId, pBackend->refId);
TAOS_UNUSED(taosReleaseRef(taskDbWrapperId, pBackend->refId));
}
void taskDbSetClearFileFlag(void* pTaskDb) {
@ -2491,24 +2487,24 @@ void taskDbInitChkpOpt(STaskDbWrapper* pTaskDb) {
pTaskDb->chkpId = -1;
pTaskDb->chkpCap = 4;
pTaskDb->chkpSaved = taosArrayInit(4, sizeof(int64_t));
(void)taskDbLoadChkpInfo(pTaskDb);
TAOS_UNUSED(taskDbLoadChkpInfo(pTaskDb));
pTaskDb->chkpInUse = taosArrayInit(4, sizeof(int64_t));
(void)taosThreadRwlockInit(&pTaskDb->chkpDirLock, NULL);
TAOS_UNUSED(taosThreadRwlockInit(&pTaskDb->chkpDirLock, NULL));
}
void taskDbRefChkp(STaskDbWrapper* pTaskDb, int64_t chkp) {
(void)taosThreadRwlockWrlock(&pTaskDb->chkpDirLock);
TAOS_UNUSED(taosThreadRwlockWrlock(&pTaskDb->chkpDirLock));
if (taosArrayPush(pTaskDb->chkpInUse, &chkp) == NULL) {
stError("failed to push chkp: %" PRIi64 " into inuse", chkp);
}
taosArraySort(pTaskDb->chkpInUse, chkpIdComp);
(void)taosThreadRwlockUnlock(&pTaskDb->chkpDirLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&pTaskDb->chkpDirLock));
}
void taskDbUnRefChkp(STaskDbWrapper* pTaskDb, int64_t chkp) {
(void)taosThreadRwlockWrlock(&pTaskDb->chkpDirLock);
TAOS_UNUSED(taosThreadRwlockWrlock(&pTaskDb->chkpDirLock));
int32_t size = taosArrayGetSize(pTaskDb->chkpInUse);
for (int i = 0; i < size; i++) {
int64_t* p = taosArrayGet(pTaskDb->chkpInUse, i);
@ -2517,13 +2513,13 @@ void taskDbUnRefChkp(STaskDbWrapper* pTaskDb, int64_t chkp) {
break;
}
}
(void)taosThreadRwlockUnlock(&pTaskDb->chkpDirLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&pTaskDb->chkpDirLock));
}
void taskDbDestroyChkpOpt(STaskDbWrapper* pTaskDb) {
taosArrayDestroy(pTaskDb->chkpSaved);
taosArrayDestroy(pTaskDb->chkpInUse);
(void)taosThreadRwlockDestroy(&pTaskDb->chkpDirLock);
TAOS_UNUSED(taosThreadRwlockDestroy(&pTaskDb->chkpDirLock));
}
int32_t taskDbBuildFullPath(char* path, char* key, char** dbFullPath, char** stateFullPath) {
@ -2569,9 +2565,9 @@ int32_t taskDbBuildFullPath(char* path, char* key, char** dbFullPath, char** sta
void taskDbUpdateChkpId(void* pTaskDb, int64_t chkpId) {
STaskDbWrapper* p = pTaskDb;
(void)streamMutexLock(&p->mutex);
TAOS_UNUSED(streamMutexLock(&p->mutex));
p->chkpId = chkpId;
(void)streamMutexUnlock(&p->mutex);
TAOS_UNUSED(streamMutexUnlock(&p->mutex));
}
STaskDbWrapper* taskDbOpenImpl(const char* key, char* statePath, char* dbPath) {
@ -2587,7 +2583,7 @@ STaskDbWrapper* taskDbOpenImpl(const char* key, char* statePath, char* dbPath) {
pTaskDb->idstr = key ? taosStrdup(key) : NULL;
pTaskDb->path = statePath ? taosStrdup(statePath) : NULL;
(void)taosThreadMutexInit(&pTaskDb->mutex, NULL);
TAOS_UNUSED(taosThreadMutexInit(&pTaskDb->mutex, NULL));
taskDbInitChkpOpt(pTaskDb);
taskDbInitOpt(pTaskDb);
@ -2772,7 +2768,7 @@ int32_t taskDbGenChkpUploadData__rsync(STaskDbWrapper* pDb, int64_t chkpId, char
char* buf = taosMemoryCalloc(1, cap);
if (buf == NULL) {
(void)taosReleaseRef(taskDbWrapperId, refId);
TAOS_UNUSED(taosReleaseRef(taskDbWrapperId, refId));
return terrno;
}
@ -2780,7 +2776,7 @@ int32_t taskDbGenChkpUploadData__rsync(STaskDbWrapper* pDb, int64_t chkpId, char
snprintf(buf, cap, "%s%s%s%s%s%" PRId64 "", pDb->path, TD_DIRSEP, "checkpoints", TD_DIRSEP, "checkpoint", chkpId);
if (nBytes <= 0 || nBytes >= cap) {
taosMemoryFree(buf);
(void)taosReleaseRef(taskDbWrapperId, refId);
TAOS_UNUSED(taosReleaseRef(taskDbWrapperId, refId));
return TSDB_CODE_OUT_OF_RANGE;
}
@ -2791,7 +2787,7 @@ int32_t taskDbGenChkpUploadData__rsync(STaskDbWrapper* pDb, int64_t chkpId, char
taosMemoryFree(buf);
}
(void)taosReleaseRef(taskDbWrapperId, refId);
TAOS_UNUSED(taosReleaseRef(taskDbWrapperId, refId));
return code;
}
@ -3028,7 +3024,7 @@ int32_t streamStateOpenBackendCf(void* backend, char* name, char** cfs, int32_t
inst->dbOpt = handle->dbOpt;
rocksdb_writeoptions_disable_WAL(inst->wOpt, 1);
(void)taosHashPut(handle->cfInst, idstr, strlen(idstr) + 1, &inst, sizeof(void*));
TAOS_UNUSED(taosHashPut(handle->cfInst, idstr, strlen(idstr) + 1, &inst, sizeof(void*)));
} else {
inst = *pInst;
}
@ -3269,9 +3265,9 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfKe
break; \
} \
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend; \
(void)atomic_add_fetch_64(&wrapper->dataWritten, 1); \
TAOS_UNUSED(atomic_add_fetch_64(&wrapper->dataWritten, 1)); \
char toString[128] = {0}; \
if (stDebugFlag & DEBUG_TRACE) (void)(ginitDict[i].toStrFunc((void*)key, toString)); \
if (stDebugFlag & DEBUG_TRACE) TAOS_UNUSED((ginitDict[i].toStrFunc((void*)key, toString))); \
int32_t klen = ginitDict[i].enFunc((void*)key, buf); \
rocksdb_column_family_handle_t* pHandle = ((rocksdb_column_family_handle_t**)wrapper->pCf)[ginitDict[i].idx]; \
rocksdb_writeoptions_t* opts = wrapper->writeOpt; \
@ -3303,7 +3299,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfKe
} \
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend; \
char toString[128] = {0}; \
if (stDebugFlag & DEBUG_TRACE) (void)(ginitDict[i].toStrFunc((void*)key, toString)); \
if (stDebugFlag & DEBUG_TRACE) TAOS_UNUSED((ginitDict[i].toStrFunc((void*)key, toString))); \
int32_t klen = ginitDict[i].enFunc((void*)key, buf); \
rocksdb_column_family_handle_t* pHandle = ((rocksdb_column_family_handle_t**)wrapper->pCf)[ginitDict[i].idx]; \
rocksdb_t* db = wrapper->db; \
@ -3346,9 +3342,9 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfKe
break; \
} \
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend; \
(void)atomic_add_fetch_64(&wrapper->dataWritten, 1); \
TAOS_UNUSED(atomic_add_fetch_64(&wrapper->dataWritten, 1)); \
char toString[128] = {0}; \
if (stDebugFlag & DEBUG_TRACE) (void)(ginitDict[i].toStrFunc((void*)key, toString)); \
if (stDebugFlag & DEBUG_TRACE) TAOS_UNUSED(ginitDict[i].toStrFunc((void*)key, toString)); \
int32_t klen = ginitDict[i].enFunc((void*)key, buf); \
rocksdb_column_family_handle_t* pHandle = ((rocksdb_column_family_handle_t**)wrapper->pCf)[ginitDict[i].idx]; \
rocksdb_t* db = wrapper->db; \
@ -3387,7 +3383,7 @@ int32_t streamStateClear_rocksdb(SStreamState* pState) {
stDebug("streamStateClear_rocksdb");
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
(void)atomic_add_fetch_64(&wrapper->dataWritten, 1);
TAOS_UNUSED(atomic_add_fetch_64(&wrapper->dataWritten, 1));
char sKeyStr[128] = {0};
char eKeyStr[128] = {0};
@ -3403,8 +3399,8 @@ int32_t streamStateClear_rocksdb(SStreamState* pState) {
if (err != NULL) {
char toStringStart[128] = {0};
char toStringEnd[128] = {0};
(void)stateKeyToString(&sKey, toStringStart);
(void)stateKeyToString(&eKey, toStringEnd);
TAOS_UNUSED(stateKeyToString(&sKey, toStringStart));
TAOS_UNUSED(stateKeyToString(&eKey, toStringEnd));
stWarn("failed to delete range cf(state) start: %s, end:%s, reason:%s", toStringStart, toStringEnd, err);
taosMemoryFree(err);
@ -3482,7 +3478,7 @@ int32_t streamStateGetKVByCur_rocksdb(SStreamStateCur* pCur, SWinKey* pKey, cons
if (rocksdb_iter_valid(pCur->iter) && !iterValueIsStale(pCur->iter)) {
size_t tlen;
char* keyStr = (char*)rocksdb_iter_key(pCur->iter, &tlen);
(void)stateKeyDecode((void*)pKtmp, keyStr);
TAOS_UNUSED(stateKeyDecode((void*)pKtmp, keyStr));
if (pKtmp->opNum != pCur->number) {
return -1;
}
@ -3535,7 +3531,7 @@ SStreamStateCur* streamStateSeekKeyNext_rocksdb(SStreamState* pState, const SWin
SStateKey curKey;
size_t kLen;
char* keyStr = (char*)rocksdb_iter_key(pCur->iter, &kLen);
(void)stateKeyDecode((void*)&curKey, keyStr);
TAOS_UNUSED(stateKeyDecode((void*)&curKey, keyStr));
if (stateKeyCmpr(&sKey, sizeof(sKey), &curKey, sizeof(curKey)) > 0) {
return pCur;
}
@ -3557,7 +3553,7 @@ SStreamStateCur* streamStateSeekToLast_rocksdb(SStreamState* pState) {
{
char tbuf[256] = {0};
(void)stateKeyToString((void*)&maxStateKey, tbuf);
TAOS_UNUSED(stateKeyToString((void*)&maxStateKey, tbuf));
stDebug("seek to last:%s", tbuf);
}
@ -3607,7 +3603,7 @@ SStreamStateCur* streamStateGetCur_rocksdb(SStreamState* pState, const SWinKey*
SStateKey curKey;
size_t kLen = 0;
char* keyStr = (char*)rocksdb_iter_key(pCur->iter, &kLen);
(void)stateKeyDecode((void*)&curKey, keyStr);
TAOS_UNUSED(stateKeyDecode((void*)&curKey, keyStr));
if (stateKeyCmpr(&sKey, sizeof(sKey), &curKey, sizeof(curKey)) == 0) {
pCur->number = pState->number;
@ -3755,7 +3751,7 @@ SStreamStateCur* streamStateSessionSeekKeyCurrentPrev_rocksdb(SStreamState* pSta
size_t klen;
const char* iKey = rocksdb_iter_key(pCur->iter, &klen);
SStateSessionKey curKey = {0};
(void)stateSessionKeyDecode(&curKey, (char*)iKey);
TAOS_UNUSED(stateSessionKeyDecode(&curKey, (char*)iKey));
if (stateSessionKeyCmpr(&sKey, sizeof(sKey), &curKey, sizeof(curKey)) >= 0) return pCur;
rocksdb_iter_prev(pCur->iter);
@ -3792,7 +3788,7 @@ SStreamStateCur* streamStateSessionSeekKeyCurrentNext_rocksdb(SStreamState* pSta
size_t klen;
const char* iKey = rocksdb_iter_key(pCur->iter, &klen);
SStateSessionKey curKey = {0};
(void)stateSessionKeyDecode(&curKey, (char*)iKey);
TAOS_UNUSED(stateSessionKeyDecode(&curKey, (char*)iKey));
if (stateSessionKeyCmpr(&sKey, sizeof(sKey), &curKey, sizeof(curKey)) <= 0) return pCur;
rocksdb_iter_next(pCur->iter);
@ -3832,7 +3828,7 @@ SStreamStateCur* streamStateSessionSeekKeyNext_rocksdb(SStreamState* pState, con
size_t klen;
const char* iKey = rocksdb_iter_key(pCur->iter, &klen);
SStateSessionKey curKey = {0};
(void)stateSessionKeyDecode(&curKey, (char*)iKey);
TAOS_UNUSED(stateSessionKeyDecode(&curKey, (char*)iKey));
if (stateSessionKeyCmpr(&sKey, sizeof(sKey), &curKey, sizeof(curKey)) < 0) return pCur;
rocksdb_iter_next(pCur->iter);
@ -3872,7 +3868,7 @@ SStreamStateCur* streamStateSessionSeekKeyPrev_rocksdb(SStreamState* pState, con
size_t klen;
const char* iKey = rocksdb_iter_key(pCur->iter, &klen);
SStateSessionKey curKey = {0};
(void)stateSessionKeyDecode(&curKey, (char*)iKey);
TAOS_UNUSED(stateSessionKeyDecode(&curKey, (char*)iKey));
if (stateSessionKeyCmpr(&sKey, sizeof(sKey), &curKey, sizeof(curKey)) > 0) return pCur;
rocksdb_iter_prev(pCur->iter);
@ -3894,7 +3890,7 @@ int32_t streamStateSessionGetKVByCur_rocksdb(SStreamStateCur* pCur, SSessionKey*
return -1;
}
const char* curKey = rocksdb_iter_key(pCur->iter, (size_t*)&kLen);
(void)stateSessionKeyDecode((void*)&ktmp, (char*)curKey);
TAOS_UNUSED(stateSessionKeyDecode((void*)&ktmp, (char*)curKey));
if (pVal != NULL) *pVal = NULL;
if (pVLen != NULL) *pVLen = 0;
@ -3973,7 +3969,7 @@ SStreamStateCur* streamStateFillGetCur_rocksdb(SStreamState* pState, const SWinK
size_t kLen;
SWinKey curKey;
char* keyStr = (char*)rocksdb_iter_key(pCur->iter, &kLen);
(void)winKeyDecode((void*)&curKey, keyStr);
TAOS_UNUSED(winKeyDecode((void*)&curKey, keyStr));
if (winKeyCmpr(key, sizeof(*key), &curKey, sizeof(curKey)) == 0) {
return pCur;
}
@ -3992,7 +3988,7 @@ int32_t streamStateFillGetKVByCur_rocksdb(SStreamStateCur* pCur, SWinKey* pKey,
}
size_t klen, vlen;
char* keyStr = (char*)rocksdb_iter_key(pCur->iter, &klen);
(void)winKeyDecode(&winKey, keyStr);
TAOS_UNUSED(winKeyDecode(&winKey, keyStr));
const char* valStr = rocksdb_iter_value(pCur->iter, &vlen);
int32_t len = valueDecode((void*)valStr, vlen, NULL, (char**)pVal);
@ -4033,7 +4029,7 @@ SStreamStateCur* streamStateFillSeekKeyNext_rocksdb(SStreamState* pState, const
SWinKey curKey;
size_t kLen = 0;
char* keyStr = (char*)rocksdb_iter_key(pCur->iter, &kLen);
(void)winKeyDecode((void*)&curKey, keyStr);
TAOS_UNUSED(winKeyDecode((void*)&curKey, keyStr));
if (winKeyCmpr(key, sizeof(*key), &curKey, sizeof(curKey)) < 0) {
return pCur;
}
@ -4070,7 +4066,7 @@ SStreamStateCur* streamStateFillSeekKeyPrev_rocksdb(SStreamState* pState, const
SWinKey curKey;
size_t kLen = 0;
char* keyStr = (char*)rocksdb_iter_key(pCur->iter, &kLen);
(void)winKeyDecode((void*)&curKey, keyStr);
TAOS_UNUSED(winKeyDecode((void*)&curKey, keyStr));
if (winKeyCmpr(key, sizeof(*key), &curKey, sizeof(curKey)) > 0) {
return pCur;
}
@ -4206,7 +4202,7 @@ void streamStateSessionClear_rocksdb(SStreamState* pState) {
if (code == 0 && size > 0) {
memset(buf, 0, size);
// refactor later
(void)streamStateSessionPut_rocksdb(pState, &delKey, buf, size);
TAOS_UNUSED(streamStateSessionPut_rocksdb(pState, &delKey, buf, size));
} else {
taosMemoryFreeClear(buf);
break;
@ -4419,7 +4415,7 @@ void streamStateDestroyBatch(void* pBatch) { rocksdb_writebatch_destroy((rock
int32_t streamStatePutBatch(SStreamState* pState, const char* cfKeyName, rocksdb_writebatch_t* pBatch, void* key,
void* val, int32_t vlen, int64_t ttl) {
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
(void)atomic_add_fetch_64(&wrapper->dataWritten, 1);
TAOS_UNUSED(atomic_add_fetch_64(&wrapper->dataWritten, 1));
int i = streamStateGetCfIdx(pState, cfKeyName);
if (i < 0) {
@ -4439,7 +4435,7 @@ int32_t streamStatePutBatch(SStreamState* pState, const char* cfKeyName, rocksdb
{
char tbuf[256] = {0};
(void)(ginitDict[i].toStrFunc((void*)key, tbuf));
TAOS_UNUSED(ginitDict[i].toStrFunc((void*)key, tbuf));
stTrace("streamState str: %s succ to write to %s_%s, len: %d", tbuf, wrapper->idstr, ginitDict[i].key, vlen);
}
return 0;
@ -4454,7 +4450,7 @@ int32_t streamStatePutBatchOptimize(SStreamState* pState, int32_t cfIdx, rocksdb
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
(void)atomic_add_fetch_64(&wrapper->dataWritten, 1);
TAOS_UNUSED(atomic_add_fetch_64(&wrapper->dataWritten, 1));
rocksdb_column_family_handle_t* pCf = wrapper->pCf[ginitDict[cfIdx].idx];
rocksdb_writebatch_put_cf((rocksdb_writebatch_t*)pBatch, pCf, buf, (size_t)klen, ttlV, (size_t)ttlVLen);
@ -4465,7 +4461,7 @@ int32_t streamStatePutBatchOptimize(SStreamState* pState, int32_t cfIdx, rocksdb
{
char tbuf[256] = {0};
(void)(ginitDict[cfIdx].toStrFunc((void*)key, tbuf));
TAOS_UNUSED(ginitDict[cfIdx].toStrFunc((void*)key, tbuf));
stTrace("streamState str: %s succ to write to %s_%s", tbuf, wrapper->idstr, ginitDict[cfIdx].key);
}
return 0;
@ -4473,7 +4469,7 @@ int32_t streamStatePutBatchOptimize(SStreamState* pState, int32_t cfIdx, rocksdb
int32_t streamStatePutBatch_rocksdb(SStreamState* pState, void* pBatch) {
char* err = NULL;
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
(void)atomic_add_fetch_64(&wrapper->dataWritten, 1);
TAOS_UNUSED(atomic_add_fetch_64(&wrapper->dataWritten, 1));
rocksdb_write(wrapper->db, wrapper->writeOpt, (rocksdb_writebatch_t*)pBatch, &err);
if (err != NULL) {
stError("streamState failed to write batch, err:%s", err);
@ -4562,7 +4558,7 @@ int32_t compareHashTableImpl(SHashObj* p1, SHashObj* p2, SArray* diff) {
if (fname == NULL) {
return terrno;
}
(void)strncpy(fname, name, len);
TAOS_UNUSED(strncpy(fname, name, len));
if (taosArrayPush(diff, &fname) == NULL) {
taosMemoryFree(fname);
return terrno;
@ -4642,7 +4638,7 @@ void dbChkpDebugInfo(SDbChkp* pDb) {
int32_t dbChkpGetDelta(SDbChkp* p, int64_t chkpId, SArray* list) {
int32_t code = 0;
int32_t nBytes;
(void)taosThreadRwlockWrlock(&p->rwLock);
TAOS_UNUSED(taosThreadRwlockWrlock(&p->rwLock));
p->preCkptId = p->curChkpId;
p->curChkpId = chkpId;
@ -4660,7 +4656,7 @@ int32_t dbChkpGetDelta(SDbChkp* p, int64_t chkpId, SArray* list) {
nBytes =
snprintf(p->buf, p->len, "%s%s%s%scheckpoint%" PRId64 "", p->path, TD_DIRSEP, "checkpoints", TD_DIRSEP, chkpId);
if (nBytes <= 0 || nBytes >= p->len) {
(void)taosThreadRwlockUnlock(&p->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&p->rwLock));
return TSDB_CODE_OUT_OF_RANGE;
}
@ -4670,7 +4666,7 @@ int32_t dbChkpGetDelta(SDbChkp* p, int64_t chkpId, SArray* list) {
TdDirPtr pDir = taosOpenDir(p->buf);
if (pDir == NULL) {
(void)taosThreadRwlockUnlock(&p->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&p->rwLock));
return terrno;
}
@ -4706,9 +4702,9 @@ int32_t dbChkpGetDelta(SDbChkp* p, int64_t chkpId, SArray* list) {
continue;
}
}
(void)taosCloseDir(&pDir);
TAOS_UNUSED(taosCloseDir(&pDir));
if (code != 0) {
(void)taosThreadRwlockUnlock(&p->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&p->rwLock));
return code;
}
@ -4720,14 +4716,14 @@ int32_t dbChkpGetDelta(SDbChkp* p, int64_t chkpId, SArray* list) {
if (name != NULL && !isBkdDataMeta(name, len)) {
char* fname = taosMemoryCalloc(1, len + 1);
if (fname == NULL) {
(void)taosThreadRwlockUnlock(&p->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&p->rwLock));
return terrno;
}
(void)strncpy(fname, name, len);
TAOS_UNUSED(strncpy(fname, name, len));
if (taosArrayPush(p->pAdd, &fname) == NULL) {
taosMemoryFree(fname);
(void)taosThreadRwlockUnlock(&p->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&p->rwLock));
return terrno;
}
}
@ -4761,7 +4757,7 @@ int32_t dbChkpGetDelta(SDbChkp* p, int64_t chkpId, SArray* list) {
p->idx = 1 - p->idx;
(void)taosThreadRwlockUnlock(&p->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&p->rwLock));
return code;
}
@ -4819,7 +4815,7 @@ int32_t dbChkpCreate(char* path, int64_t initChkpId, SDbChkp** ppChkp) {
}
p->update = 0;
(void)taosThreadRwlockInit(&p->rwLock, NULL);
TAOS_UNUSED(taosThreadRwlockInit(&p->rwLock, NULL));
SArray* list = NULL;
code = dbChkpGetDelta(p, initChkpId, list);
@ -4860,7 +4856,7 @@ int32_t dbChkpDumpTo(SDbChkp* p, char* dname, SArray* list) {
static char* chkpMeta = "META";
int32_t code = 0;
(void)taosThreadRwlockRdlock(&p->rwLock);
TAOS_UNUSED(taosThreadRwlockRdlock(&p->rwLock));
int32_t cap = p->len + 128;
@ -5003,7 +4999,7 @@ int32_t dbChkpDumpTo(SDbChkp* p, char* dname, SArray* list) {
if (nBytes <= 0 || nBytes >= sizeof(content)) {
code = TSDB_CODE_OUT_OF_RANGE;
stError("chkp failed to format meta file: %s, reason: invalid msg", dstDir);
(void)taosCloseFile(&pFile);
TAOS_UNUSED(taosCloseFile(&pFile));
goto _ERROR;
}
@ -5011,10 +5007,10 @@ int32_t dbChkpDumpTo(SDbChkp* p, char* dname, SArray* list) {
if (nBytes != strlen(content)) {
code = terrno;
stError("chkp failed to write meta file: %s,reason:%s", dstDir, tstrerror(code));
(void)taosCloseFile(&pFile);
TAOS_UNUSED(taosCloseFile(&pFile));
goto _ERROR;
}
(void)taosCloseFile(&pFile);
TAOS_UNUSED(taosCloseFile(&pFile));
// clear delta data buf
taosArrayClearP(p->pAdd, taosMemoryFree);
@ -5023,7 +5019,7 @@ int32_t dbChkpDumpTo(SDbChkp* p, char* dname, SArray* list) {
_ERROR:
taosMemoryFree(buffer);
(void)taosThreadRwlockUnlock(&p->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&p->rwLock));
return code;
}
@ -5068,7 +5064,7 @@ void bkdMgtDestroy(SBkdMgt* bm) {
pIter = taosHashIterate(bm->pDbChkpTbl, pIter);
}
(void)taosThreadRwlockDestroy(&bm->rwLock);
TAOS_UNUSED(taosThreadRwlockDestroy(&bm->rwLock));
taosMemoryFree(bm->path);
taosHashCleanup(bm->pDbChkpTbl);
@ -5076,7 +5072,7 @@ void bkdMgtDestroy(SBkdMgt* bm) {
}
int32_t bkdMgtGetDelta(SBkdMgt* bm, char* taskId, int64_t chkpId, SArray* list, char* dname) {
int32_t code = 0;
(void)taosThreadRwlockWrlock(&bm->rwLock);
TAOS_UNUSED(taosThreadRwlockWrlock(&bm->rwLock));
SDbChkp** ppChkp = taosHashGet(bm->pDbChkpTbl, taskId, strlen(taskId));
SDbChkp* pChkp = ppChkp != NULL ? *ppChkp : NULL;
@ -5084,14 +5080,14 @@ int32_t bkdMgtGetDelta(SBkdMgt* bm, char* taskId, int64_t chkpId, SArray* list,
int32_t cap = strlen(bm->path) + 64;
char* path = taosMemoryCalloc(1, cap);
if (path == NULL) {
(void)taosThreadRwlockUnlock(&bm->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&bm->rwLock));
return terrno;
}
int32_t nBytes = snprintf(path, cap, "%s%s%s", bm->path, TD_DIRSEP, taskId);
if (nBytes <= 0 || nBytes >= cap) {
taosMemoryFree(path);
(void)taosThreadRwlockUnlock(&bm->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&bm->rwLock));
code = TSDB_CODE_OUT_OF_RANGE;
return code;
}
@ -5100,20 +5096,20 @@ int32_t bkdMgtGetDelta(SBkdMgt* bm, char* taskId, int64_t chkpId, SArray* list,
code = dbChkpCreate(path, chkpId, &p);
if (code != 0) {
taosMemoryFree(path);
(void)taosThreadRwlockUnlock(&bm->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&bm->rwLock));
return code;
}
if (taosHashPut(bm->pDbChkpTbl, taskId, strlen(taskId), &p, sizeof(void*)) != 0) {
dbChkpDestroy(p);
(void)taosThreadRwlockUnlock(&bm->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&bm->rwLock));
code = terrno;
return code;
}
pChkp = p;
code = dbChkpDumpTo(pChkp, dname, list);
(void)taosThreadRwlockUnlock(&bm->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&bm->rwLock));
return code;
} else {
code = dbChkpGetDelta(pChkp, chkpId, NULL);
@ -5122,7 +5118,7 @@ int32_t bkdMgtGetDelta(SBkdMgt* bm, char* taskId, int64_t chkpId, SArray* list,
}
}
(void)taosThreadRwlockUnlock(&bm->rwLock);
TAOS_UNUSED(taosThreadRwlockUnlock(&bm->rwLock));
return code;
}

46
source/libs/transport/src/thttp.c
View File

@ -236,7 +236,7 @@ static FORCE_INLINE int32_t taosBuildDstAddr(const char* server, uint16_t port,
static void* httpThread(void* arg) {
SHttpModule* http = (SHttpModule*)arg;
setThreadName("http-cli-send-thread");
(void)uv_run(http->loop, UV_RUN_DEFAULT);
TAOS_UNUSED(uv_run(http->loop, UV_RUN_DEFAULT));
return NULL;
}
@ -352,7 +352,7 @@ static void httpAsyncCb(uv_async_t* handle) {
static int32_t BATCH_SIZE = 20;
int32_t count = 0;
(void)taosThreadMutexLock(&item->mtx);
TAOS_UNUSED(taosThreadMutexLock(&item->mtx));
httpMayDiscardMsg(http, item);
while (!QUEUE_IS_EMPTY(&item->qmsg) && count++ < BATCH_SIZE) {
@ -360,7 +360,7 @@ static void httpAsyncCb(uv_async_t* handle) {
QUEUE_REMOVE(h);
QUEUE_PUSH(&wq, h);
}
(void)taosThreadMutexUnlock(&item->mtx);
TAOS_UNUSED(taosThreadMutexUnlock(&item->mtx));
httpTrace(&wq);
@ -393,7 +393,7 @@ static FORCE_INLINE void clientCloseCb(uv_handle_t* handle) {
SHttpModule* http = taosAcquireRef(httpRefMgt, cli->chanId);
if (http != NULL) {
http->connNum -= 1;
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
}
destroyHttpClient(cli);
@ -453,7 +453,7 @@ static void clientConnCb(uv_connect_t* req, int32_t status) {
if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
}
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
http->connNum += 1;
@ -468,7 +468,7 @@ static void clientConnCb(uv_connect_t* req, int32_t status) {
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
}
}
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
}
int32_t httpSendQuit(SHttpModule* http, int64_t chanId) {
@ -514,7 +514,7 @@ static void httpHandleQuit(SHttpMsg* msg) {
SHttpModule* http = taosAcquireRef(httpRefMgt, chanId);
if (http == NULL) return;
uv_walk(http->loop, httpWalkCb, NULL);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
}
static bool httpFailFastShoudIgnoreMsg(SHashObj* pTable, char* server, int16_t port) {
@ -536,15 +536,15 @@ static bool httpFailFastShoudIgnoreMsg(SHashObj* pTable, char* server, int16_t p
}
static void httpFailFastMayUpdate(SHashObj* pTable, char* server, int16_t port, int8_t succ) {
int32_t code = 0;
char buf[256] = {0};
char buf[256] = {0};
sprintf(buf, "%s:%d", server, port);
if (succ) {
(void)taosHashRemove(pTable, buf, strlen(buf));
TAOS_UNUSED(taosHashRemove(pTable, buf, strlen(buf)));
} else {
int32_t st = taosGetTimestampSec();
if ((code = taosHashPut(pTable, buf, strlen(buf), &st, sizeof(st))) != 0) {
tError("http-report failed to update conn status, dst:%s, reason:%s", buf, tstrerror(code));
tError("http-report failed to update conn status, dst:%s, reason:%s", buf, tstrerror(code));
}
}
return;
@ -628,7 +628,7 @@ static void httpHandleReq(SHttpMsg* msg) {
tError("http-report failed to alloc read buf, dst:%s:%d, chanId:%" PRId64 ", seq:%" PRId64 ",reason:%s", cli->addr,
cli->port, chanId, cli->seq, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
destroyHttpClient(cli);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
@ -637,7 +637,7 @@ static void httpHandleReq(SHttpMsg* msg) {
tError("http-report failed to init socket handle, dst:%s:%d, chanId:%" PRId64 ", seq:%" PRId64 ", reason:%s",
cli->addr, cli->port, chanId, cli->seq, uv_strerror(err));
destroyHttpClient(cli);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
@ -647,7 +647,7 @@ static void httpHandleReq(SHttpMsg* msg) {
tError("http-report failed to open socket, dst:%s:%d, chanId:%" PRId64 ", seq:%" PRId64 ", reason:%s", cli->addr,
cli->port, chanId, cli->seq, tstrerror(terrno));
destroyHttpClient(cli);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
@ -657,7 +657,7 @@ static void httpHandleReq(SHttpMsg* msg) {
cli->port, chanId, cli->seq, uv_strerror(ret));
destroyHttpClient(cli);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
@ -668,7 +668,7 @@ static void httpHandleReq(SHttpMsg* msg) {
httpFailFastMayUpdate(http->connStatusTable, cli->addr, cli->port, 0);
destroyHttpClient(cli);
}
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
END:
@ -678,7 +678,7 @@ END:
}
httpDestroyMsg(msg);
taosMemoryFree(header);
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
}
static void httpModuleDestroy(SHttpModule* http) {
@ -689,7 +689,7 @@ static void httpModuleDestroy(SHttpModule* http) {
transAsyncPoolDestroy(http->asyncPool);
}
if (http->loop) {
(void)uv_loop_close(http->loop);
TAOS_UNUSED(uv_loop_close(http->loop));
taosMemoryFree(http->loop);
}
@ -754,7 +754,7 @@ int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, c
int32_t taosSendHttpReportWithQID(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag, const char* qid) {
(void)taosThreadOnce(&transHttpInit, transHttpEnvInit);
TAOS_UNUSED(taosThreadOnce(&transHttpInit, transHttpEnvInit));
return taosSendHttpReportImplByChan(server, uri, port, pCont, contLen, flag, httpDefaultChanId, qid);
}
@ -825,7 +825,7 @@ _ERROR:
return code;
}
int64_t taosInitHttpChan() {
(void)taosThreadOnce(&transHttpInit, transHttpEnvInit);
TAOS_UNUSED(taosThreadOnce(&transHttpInit, transHttpEnvInit));
return transInitHttpChanImpl();
}
@ -844,14 +844,14 @@ void taosDestroyHttpChan(int64_t chanId) {
ret = httpSendQuit(load, chanId);
if (ret != 0) {
tDebug("http-report already destroyed, chanId %" PRId64 ",reason:%s", chanId, tstrerror(ret));
(void)taosReleaseRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
return;
}
(void)taosThreadJoin(load->thread, NULL);
TAOS_UNUSED(taosThreadJoin(load->thread, NULL));
httpModuleDestroy(load);
(void)taosReleaseRef(httpRefMgt, chanId);
(void)taosRemoveRef(httpRefMgt, chanId);
TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId));
TAOS_UNUSED(taosRemoveRef(httpRefMgt, chanId));
}

205
source/libs/transport/src/transCli.c
View File

@ -233,11 +233,11 @@ static void destroyThrdObj(SCliThrd* pThrd);
int32_t cliSendQuit(SCliThrd* thrd);
static void cliWalkCb(uv_handle_t* handle, void* arg);
#define CLI_RELEASE_UV(loop) \
do { \
(void)uv_walk(loop, cliWalkCb, NULL); \
(void)uv_run(loop, UV_RUN_DEFAULT); \
(void)uv_loop_close(loop); \
#define CLI_RELEASE_UV(loop) \
do { \
TAOS_UNUSED(uv_walk(loop, cliWalkCb, NULL)); \
TAOS_UNUSED(uv_run(loop, UV_RUN_DEFAULT)); \
TAOS_UNUSED(uv_loop_close(loop)); \
} while (0);
// snprintf may cause performance problem
@ -247,7 +247,7 @@ static void cliWalkCb(uv_handle_t* handle, void* arg);
int16_t len = strlen(ip); \
if (ip != NULL) memcpy(t, ip, len); \
t[len] = ':'; \
(void)titoa(port, 10, &t[len + 1]); \
TAOS_UNUSED(titoa(port, 10, &t[len + 1])); \
} while (0)
#define CONN_PERSIST_TIME(para) ((para) <= 90000 ? 90000 : (para))
@ -336,7 +336,7 @@ void cliResetTimer(SCliThrd* pThrd, SCliConn* conn) {
if (conn->timer) {
if (uv_is_active((uv_handle_t*)conn->timer)) {
tDebug("%s conn %p stop timer", CONN_GET_INST_LABEL(conn), conn);
(void)uv_timer_stop(conn->timer);
TAOS_UNUSED(uv_timer_stop(conn->timer));
}
if (taosArrayPush(pThrd->timerList, &conn->timer) == NULL) {
tError("failed to push timer %p to list, reason:%s", conn->timer, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
@ -375,15 +375,15 @@ bool cliConnSendSeqMsg(int64_t refId, SCliConn* conn) {
taosWUnLockLatch(&exh->latch);
SCliMsg* t = QUEUE_DATA(h, SCliMsg, seqq);
transCtxMerge(&conn->ctx, &t->ctx->appCtx);
(void)transQueuePush(&conn->cliMsgs, t);
TAOS_UNUSED(transQueuePush(&conn->cliMsgs, t));
tDebug("pop from conn %p, refId: %" PRId64 "", conn, refId);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
cliSend(conn);
return true;
}
taosWUnLockLatch(&exh->latch);
tDebug("empty conn %p, refId: %" PRId64 "", conn, refId);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
return false;
}
@ -496,7 +496,7 @@ void cliHandleResp(SCliConn* conn) {
return addConnToPool(pThrd->pool, conn);
}
(void)uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb);
TAOS_UNUSED(uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb));
}
static void cliDestroyMsgInExhandle(int64_t refId) {
if (refId == 0) return;
@ -510,7 +510,7 @@ static void cliDestroyMsgInExhandle(int64_t refId) {
destroyCmsg(t);
}
taosWUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
}
}
@ -598,7 +598,7 @@ void cliConnTimeout(uv_timer_t* handle) {
tTrace("%s conn %p conn timeout, ref:%d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn));
(void)uv_timer_stop(handle);
TAOS_UNUSED(uv_timer_stop(handle));
handle->data = NULL;
cliResetTimer(pThrd, conn);
@ -610,7 +610,7 @@ void cliReadTimeoutCb(uv_timer_t* handle) {
// set up timeout cb
SCliConn* conn = handle->data;
tTrace("%s conn %p timeout, ref:%d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn));
(void)uv_read_stop(conn->stream);
TAOS_UNUSED(uv_read_stop(conn->stream));
cliHandleExceptImpl(conn, TSDB_CODE_RPC_TIMEOUT);
}
@ -847,7 +847,7 @@ static void addConnToPool(void* pool, SCliConn* conn) {
pMsg->ctx->task = NULL;
transCtxMerge(&conn->ctx, &pMsg->ctx->appCtx);
(void)transQueuePush(&conn->cliMsgs, pMsg);
TAOS_UNUSED(transQueuePush(&conn->cliMsgs, pMsg));
conn->status = ConnNormal;
cliSend(conn);
@ -871,8 +871,8 @@ static void addConnToPool(void* pool, SCliConn* conn) {
}
static int32_t allocConnRef(SCliConn* conn, bool update) {
if (update) {
(void)transReleaseExHandle(transGetRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), conn->refId));
conn->refId = -1;
}
@ -907,8 +907,8 @@ static int32_t allocConnRef(SCliConn* conn, bool update) {
static int32_t specifyConnRef(SCliConn* conn, bool update, int64_t handle) {
if (update) {
(void)transReleaseExHandle(transGetRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), conn->refId));
conn->refId = -1;
}
SExHandle* exh = transAcquireExHandle(transGetRefMgt(), handle);
@ -924,7 +924,7 @@ static int32_t specifyConnRef(SCliConn* conn, bool update, int64_t handle) {
tDebug("conn %p specified by %" PRId64 "", conn, handle);
(void)transReleaseExHandle(transGetRefMgt(), handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
return 0;
}
@ -1004,7 +1004,7 @@ static int32_t cliCreateConn(SCliThrd* pThrd, SCliConn** pCliConn) {
}
tDebug("no available timer, create a timer %p", timer);
(void)uv_timer_init(pThrd->loop, timer);
TAOS_UNUSED(uv_timer_init(pThrd->loop, timer));
}
timer->data = conn;
@ -1022,7 +1022,7 @@ static int32_t cliCreateConn(SCliThrd* pThrd, SCliConn** pCliConn) {
conn->broken = false;
transRefCliHandle(conn);
(void)atomic_add_fetch_32(&pThrd->connCount, 1);
TAOS_UNUSED(atomic_add_fetch_32(&pThrd->connCount, 1));
TAOS_CHECK_GOTO(allocConnRef(conn, false), NULL, _failed);
@ -1032,7 +1032,7 @@ _failed:
if (conn) {
taosMemoryFree(conn->stream);
transReqQueueClear(&conn->wreqQueue);
(void)transDestroyBuffer(&conn->readBuf);
TAOS_UNUSED(transDestroyBuffer(&conn->readBuf));
transQueueDestroy(&conn->cliMsgs);
}
taosMemoryFree(conn);
@ -1059,8 +1059,8 @@ static void cliDestroyConn(SCliConn* conn, bool clear) {
}
conn->list = NULL;
(void)transReleaseExHandle(transGetRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), conn->refId));
conn->refId = -1;
if (conn->task != NULL) {
@ -1071,7 +1071,7 @@ static void cliDestroyConn(SCliConn* conn, bool clear) {
if (clear) {
if (!uv_is_closing((uv_handle_t*)conn->stream)) {
(void)uv_read_stop(conn->stream);
TAOS_UNUSED(uv_read_stop(conn->stream));
uv_close((uv_handle_t*)conn->stream, cliDestroy);
}
}
@ -1084,11 +1084,11 @@ static void cliDestroy(uv_handle_t* handle) {
SCliThrd* pThrd = conn->hostThrd;
cliResetTimer(pThrd, conn);
(void)atomic_sub_fetch_32(&pThrd->connCount, 1);
TAOS_UNUSED(atomic_sub_fetch_32(&pThrd->connCount, 1));
if (conn->refId > 0) {
(void)transReleaseExHandle(transGetRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), conn->refId));
}
taosMemoryFree(conn->dstAddr);
taosMemoryFree(conn->stream);
@ -1097,7 +1097,7 @@ static void cliDestroy(uv_handle_t* handle) {
tTrace("%s conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn);
transReqQueueClear(&conn->wreqQueue);
(void)transDestroyBuffer(&conn->readBuf);
TAOS_UNUSED(transDestroyBuffer(&conn->readBuf));
taosMemoryFree(conn);
}
@ -1106,7 +1106,7 @@ static bool cliHandleNoResp(SCliConn* conn) {
if (!transQueueEmpty(&conn->cliMsgs)) {
SCliMsg* pMsg = transQueueGet(&conn->cliMsgs, 0);
if (REQUEST_NO_RESP(&pMsg->msg)) {
(void)transQueuePop(&conn->cliMsgs);
TAOS_UNUSED(transQueuePop(&conn->cliMsgs));
destroyCmsg(pMsg);
res = true;
}
@ -1153,7 +1153,7 @@ static void cliSendCb(uv_write_t* req, int status) {
tTrace("%s conn %p no resp required", CONN_GET_INST_LABEL(pConn), pConn);
return;
}
(void)uv_read_start((uv_stream_t*)pConn->stream, cliAllocRecvBufferCb, cliRecvCb);
TAOS_UNUSED(uv_read_start((uv_stream_t*)pConn->stream, cliAllocRecvBufferCb, cliRecvCb));
}
void cliSendBatch(SCliConn* pConn) {
int32_t code = 0;
@ -1296,13 +1296,13 @@ void cliSend(SCliConn* pConn) {
if (timer == NULL) {
timer = taosMemoryCalloc(1, sizeof(uv_timer_t));
tDebug("no available timer, create a timer %p", timer);
(void)uv_timer_init(pThrd->loop, timer);
TAOS_UNUSED(uv_timer_init(pThrd->loop, timer));
}
timer->data = pConn;
pConn->timer = timer;
tGTrace("%s conn %p start timer for msg:%s", CONN_GET_INST_LABEL(pConn), pConn, TMSG_INFO(pMsg->msgType));
(void)uv_timer_start((uv_timer_t*)pConn->timer, cliReadTimeoutCb, TRANS_READ_TIMEOUT, 0);
TAOS_UNUSED(uv_timer_start((uv_timer_t*)pConn->timer, cliReadTimeoutCb, TRANS_READ_TIMEOUT, 0));
}
if (pHead->comp == 0 && pMsg->info.compressed == 0 && pConn->clientIp != pConn->serverIp) {
@ -1402,8 +1402,7 @@ static void cliHandleBatchReq(SCliBatch* pBatch, SCliThrd* pThrd) {
tTrace("%s conn %p try to connect to %s", pTransInst->label, conn, pList->dst);
int32_t fd = taosCreateSocketWithTimeout(TRANS_CONN_TIMEOUT * 10);
if (fd == -1) {
tError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn,
tstrerror(terrno));
tError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn, tstrerror(terrno));
cliHandleFastFail(conn, -1);
return;
}
@ -1428,7 +1427,7 @@ static void cliHandleBatchReq(SCliBatch* pBatch, SCliThrd* pThrd) {
cliHandleFastFail(conn, -1);
return;
}
(void)uv_timer_start(conn->timer, cliConnTimeout, TRANS_CONN_TIMEOUT, 0);
TAOS_UNUSED(uv_timer_start(conn->timer, cliConnTimeout, TRANS_CONN_TIMEOUT, 0));
return;
}
@ -1541,12 +1540,12 @@ void cliConnCb(uv_connect_t* req, int status) {
struct sockaddr peername, sockname;
int addrlen = sizeof(peername);
(void)uv_tcp_getpeername((uv_tcp_t*)pConn->stream, &peername, &addrlen);
(void)transSockInfo2Str(&peername, pConn->dst);
TAOS_UNUSED(uv_tcp_getpeername((uv_tcp_t*)pConn->stream, &peername, &addrlen));
TAOS_UNUSED(transSockInfo2Str(&peername, pConn->dst));
addrlen = sizeof(sockname);
(void)uv_tcp_getsockname((uv_tcp_t*)pConn->stream, &sockname, &addrlen);
(void)transSockInfo2Str(&sockname, pConn->src);
TAOS_UNUSED(uv_tcp_getsockname((uv_tcp_t*)pConn->stream, &sockname, &addrlen));
TAOS_UNUSED(transSockInfo2Str(&sockname, pConn->src));
struct sockaddr_in addr = *(struct sockaddr_in*)&sockname;
struct sockaddr_in saddr = *(struct sockaddr_in*)&peername;
@ -1596,8 +1595,8 @@ static void cliHandleQuit(SCliMsg* pMsg, SCliThrd* pThrd) {
tDebug("cli work thread %p start to quit", pThrd);
destroyCmsg(pMsg);
(void)destroyConnPool(pThrd);
(void)uv_walk(pThrd->loop, cliWalkCb, NULL);
TAOS_UNUSED(destroyConnPool(pThrd));
TAOS_UNUSED(uv_walk(pThrd->loop, cliWalkCb, NULL));
}
static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) {
int64_t refId = (int64_t)(pMsg->msg.info.handle);
@ -1611,8 +1610,7 @@ static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) {
taosRLockLatch(&exh->latch);
SCliConn* conn = exh->handle;
taosRUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
tDebug("%s conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn);
if (T_REF_VAL_GET(conn) == 2) {
@ -1656,7 +1654,7 @@ static void cliHandleFreeById(SCliMsg* pMsg, SCliThrd* pThrd) {
TAOS_CHECK_GOTO(TSDB_CODE_REF_INVALID_ID, NULL, _exception);
} else {
destroyCmsg(pMsg);
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
while (T_REF_VAL_GET(conn) >= 1) {
transUnrefCliHandle(conn);
@ -1666,9 +1664,9 @@ static void cliHandleFreeById(SCliMsg* pMsg, SCliThrd* pThrd) {
_exception:
tDebug("already free conn %p by id %" PRId64 "", conn, refId);
(void)transReleaseExHandle(transGetRefMgt(), refId);
(void)transReleaseExHandle(transGetRefMgt(), refId);
(void)transRemoveExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), refId));
destroyCmsg(pMsg);
}
@ -1691,7 +1689,7 @@ SCliConn* cliGetConn(SCliMsg** pMsg, SCliThrd* pThrd, bool* ignore, char* addr)
conn = getConnFromPool2(pThrd, addr, pMsg);
if (conn != NULL) specifyConnRef(conn, true, refId);
}
(void)transReleaseExHandle(transGetRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), refId));
}
return conn;
};
@ -1787,7 +1785,7 @@ static void cliMayUpdateFqdnCache(SHashObj* cache, char* dst) {
if (i > 0) {
char fqdn[TSDB_FQDN_LEN + 1] = {0};
memcpy(fqdn, dst, i);
(void)cliUpdateFqdnCache(cache, fqdn);
TAOS_UNUSED(cliUpdateFqdnCache(cache, fqdn));
}
}
@ -1824,7 +1822,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
if (ignore == true) {
// persist conn already release by server
STransMsg resp = {0};
(void)cliBuildExceptResp(pMsg, &resp);
TAOS_UNUSED(cliBuildExceptResp(pMsg, &resp));
// refactorr later
resp.info.cliVer = pTransInst->compatibilityVer;
@ -1841,14 +1839,14 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
if (conn != NULL) {
transCtxMerge(&conn->ctx, &pMsg->ctx->appCtx);
(void)transQueuePush(&conn->cliMsgs, pMsg);
TAOS_UNUSED(transQueuePush(&conn->cliMsgs, pMsg));
cliSend(conn);
} else {
code = cliCreateConn(pThrd, &conn);
if (code != 0) {
tError("%s failed to create conn, reason:%s", pTransInst->label, tstrerror(code));
STransMsg resp = {.code = code};
(void)cliBuildExceptResp(pMsg, &resp);
TAOS_UNUSED(cliBuildExceptResp(pMsg, &resp));
resp.info.cliVer = pTransInst->compatibilityVer;
if (pMsg->type != Release) {
@ -1862,7 +1860,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
if (refId != 0) specifyConnRef(conn, true, refId);
transCtxMerge(&conn->ctx, &pMsg->ctx->appCtx);
(void)transQueuePush(&conn->cliMsgs, pMsg);
TAOS_UNUSED(transQueuePush(&conn->cliMsgs, pMsg));
conn->dstAddr = taosStrdup(addr);
@ -1882,8 +1880,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
tGTrace("%s conn %p try to connect to %s", pTransInst->label, conn, conn->dstAddr);
int32_t fd = taosCreateSocketWithTimeout(TRANS_CONN_TIMEOUT * 10);
if (fd == -1) {
tGError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn,
tstrerror(terrno));
tGError("%s conn %p failed to create socket, reason:%s", transLabel(pTransInst), conn, tstrerror(terrno));
cliHandleExcept(conn, -1);
terrno = 0;
return;
@ -1911,7 +1908,7 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
cliHandleFastFail(conn, ret);
return;
}
(void)uv_timer_start(conn->timer, cliConnTimeout, TRANS_CONN_TIMEOUT, 0);
TAOS_UNUSED(uv_timer_start(conn->timer, cliConnTimeout, TRANS_CONN_TIMEOUT, 0));
}
tGTrace("%s conn %p ready", pTransInst->label, conn);
}
@ -2110,9 +2107,9 @@ static void cliAsyncCb(uv_async_t* handle) {
// batch process to avoid to lock/unlock frequently
queue wq;
(void)taosThreadMutexLock(&item->mtx);
TAOS_UNUSED(taosThreadMutexLock(&item->mtx));
QUEUE_MOVE(&item->qmsg, &wq);
(void)taosThreadMutexUnlock(&item->mtx);
TAOS_UNUSED(taosThreadMutexUnlock(&item->mtx));
int8_t supportBatch = pTransInst->supportBatch;
if (supportBatch == 0) {
@ -2162,7 +2159,7 @@ bool cliRecvReleaseReq(SCliConn* conn, STransMsgHead* pHead) {
tDebug("%s conn %p receive release request, refId:%" PRId64 ", may ignore", CONN_GET_INST_LABEL(conn), conn,
conn->refId);
(void)transClearBuffer(&conn->readBuf);
TAOS_UNUSED(transClearBuffer(&conn->readBuf));
transFreeMsg(transContFromHead((char*)pHead));
for (int i = 0; ahandle == 0 && i < transQueueSize(&conn->cliMsgs); i++) {
@ -2192,10 +2189,10 @@ static void* cliWorkThread(void* arg) {
SCliThrd* pThrd = (SCliThrd*)arg;
pThrd->pid = taosGetSelfPthreadId();
(void)strtolower(threadName, pThrd->pTransInst->label);
TAOS_UNUSED(strtolower(threadName, pThrd->pTransInst->label));
setThreadName(threadName);
(void)uv_run(pThrd->loop, UV_RUN_DEFAULT);
TAOS_UNUSED(uv_run(pThrd->loop, UV_RUN_DEFAULT));
tDebug("thread quit-thread:%08" PRId64, pThrd->pid);
return NULL;
@ -2239,7 +2236,7 @@ _err:
if (cli) {
for (int i = 0; i < cli->numOfThreads; i++) {
if (cli->pThreadObj[i]) {
(void)cliSendQuit(cli->pThreadObj[i]);
TAOS_UNUSED(cliSendQuit(cli->pThreadObj[i]));
destroyThrdObj(cli->pThreadObj[i]);
}
}
@ -2283,8 +2280,8 @@ static FORCE_INLINE void destroyCmsgAndAhandle(void* param) {
}
if (pMsg->msg.info.handle != 0) {
(void)transReleaseExHandle(transGetRefMgt(), (int64_t)pMsg->msg.info.handle);
(void)transRemoveExHandle(transGetRefMgt(), (int64_t)pMsg->msg.info.handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), (int64_t)pMsg->msg.info.handle));
TAOS_UNUSED(transRemoveExHandle(transGetRefMgt(), (int64_t)pMsg->msg.info.handle));
}
transDestroyConnCtx(pMsg->ctx);
@ -2302,7 +2299,7 @@ static int32_t createThrdObj(void* trans, SCliThrd** ppThrd) {
}
QUEUE_INIT(&pThrd->msg);
(void)taosThreadMutexInit(&pThrd->msgMtx, NULL);
TAOS_UNUSED(taosThreadMutexInit(&pThrd->msgMtx, NULL));
pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
if (pThrd->loop == NULL) {
@ -2366,7 +2363,7 @@ static int32_t createThrdObj(void* trans, SCliThrd** ppThrd) {
if (timer == NULL) {
TAOS_CHECK_GOTO(terrno, NULL, _end);
}
(void)uv_timer_init(pThrd->loop, timer);
TAOS_UNUSED(uv_timer_init(pThrd->loop, timer));
if (taosArrayPush(pThrd->timerList, &timer) == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, NULL, _end);
}
@ -2380,19 +2377,19 @@ static int32_t createThrdObj(void* trans, SCliThrd** ppThrd) {
_end:
if (pThrd) {
(void)taosThreadMutexDestroy(&pThrd->msgMtx);
TAOS_UNUSED(taosThreadMutexDestroy(&pThrd->msgMtx));
(void)uv_loop_close(pThrd->loop);
TAOS_UNUSED(uv_loop_close(pThrd->loop));
taosMemoryFree(pThrd->loop);
transAsyncPoolDestroy(pThrd->asyncPool);
for (int i = 0; i < taosArrayGetSize(pThrd->timerList); i++) {
uv_timer_t* timer = taosArrayGetP(pThrd->timerList, i);
(void)uv_timer_stop(timer);
TAOS_UNUSED(uv_timer_stop(timer));
taosMemoryFree(timer);
}
taosArrayDestroy(pThrd->timerList);
(void)destroyConnPool(pThrd);
TAOS_UNUSED(destroyConnPool(pThrd));
transDQDestroy(pThrd->delayQueue, NULL);
transDQDestroy(pThrd->timeoutQueue, NULL);
transDQDestroy(pThrd->waitConnQueue, NULL);
@ -2409,9 +2406,9 @@ static void destroyThrdObj(SCliThrd* pThrd) {
return;
}
(void)taosThreadJoin(pThrd->thread, NULL);
TAOS_UNUSED(taosThreadJoin(pThrd->thread, NULL));
CLI_RELEASE_UV(pThrd->loop);
(void)taosThreadMutexDestroy(&pThrd->msgMtx);
TAOS_UNUSED(taosThreadMutexDestroy(&pThrd->msgMtx));
TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SCliMsg, destroyCmsgWrapper, (void*)pThrd);
transAsyncPoolDestroy(pThrd->asyncPool);
@ -2478,9 +2475,9 @@ void cliWalkCb(uv_handle_t* handle, void* arg) {
if (uv_handle_get_type(handle) == UV_TIMER) {
// do nothing
} else {
(void)uv_read_stop((uv_stream_t*)handle);
TAOS_UNUSED(uv_read_stop((uv_stream_t*)handle));
}
(void)uv_close(handle, cliDestroy);
TAOS_UNUSED(uv_close(handle, cliDestroy));
}
}
@ -2518,7 +2515,7 @@ static void cliSchedMsgToDebug(SCliMsg* pMsg, char* label) {
STransConnCtx* pCtx = pMsg->ctx;
STraceId* trace = &pMsg->msg.info.traceId;
char tbuf[512] = {0};
(void)epsetToStr(&pCtx->epSet, tbuf, tListLen(tbuf));
TAOS_UNUSED(epsetToStr(&pCtx->epSet, tbuf, tListLen(tbuf)));
tGDebug("%s retry on next node,use:%s, step: %d,timeout:%" PRId64 "", label, tbuf, pCtx->retryStep,
pCtx->retryNextInterval);
return;
@ -2533,7 +2530,7 @@ static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) {
arg->param1 = pMsg;
arg->param2 = pThrd;
(void)transDQSched(pThrd->delayQueue, doDelayTask, arg, pCtx->retryNextInterval);
TAOS_UNUSED(transDQSched(pThrd->delayQueue, doDelayTask, arg, pCtx->retryNextInterval));
}
FORCE_INLINE bool cliTryExtractEpSet(STransMsg* pResp, SEpSet* dst) {
@ -2751,7 +2748,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
if (hasEpSet) {
if (rpcDebugFlag & DEBUG_TRACE) {
char tbuf[512] = {0};
(void)epsetToStr(&pCtx->epSet, tbuf, tListLen(tbuf));
TAOS_UNUSED(epsetToStr(&pCtx->epSet, tbuf, tListLen(tbuf)));
tGTrace("%s conn %p extract epset from msg", CONN_GET_INST_LABEL(pConn), pConn);
}
}
@ -2763,7 +2760,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
} else {
memcpy((char*)pCtx->pRsp, (char*)pResp, sizeof(*pResp));
}
(void)tsem_post(pCtx->pSem);
TAOS_UNUSED(tsem_post(pCtx->pSem));
pCtx->pRsp = NULL;
} else {
STransSyncMsg* pSyncMsg = taosAcquireRef(transGetSyncMsgMgt(), pCtx->syncMsgRef);
@ -2773,8 +2770,8 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
pSyncMsg->hasEpSet = 1;
epsetAssign(&pSyncMsg->epSet, &pCtx->epSet);
}
(void)tsem2_post(pSyncMsg->pSem);
(void)taosReleaseRef(transGetSyncMsgMgt(), pCtx->syncMsgRef);
TAOS_UNUSED(tsem2_post(pSyncMsg->pSem));
TAOS_UNUSED(taosReleaseRef(transGetSyncMsgMgt(), pCtx->syncMsgRef));
} else {
rpcFreeCont(pResp->pCont);
}
@ -2843,7 +2840,7 @@ static FORCE_INLINE SCliThrd* transGetWorkThrdFromHandle(STrans* trans, int64_t
pThrd = exh->pThrd;
taosWUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
return pThrd;
}
@ -2949,7 +2946,7 @@ int32_t transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, S
code = transInitMsg(shandle, pEpSet, pReq, ctx, &pCliMsg);
if (code != 0) {
taosWUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
TAOS_CHECK_GOTO(code, NULL, _exception);
}
@ -2957,13 +2954,13 @@ int32_t transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, S
taosWUnLockLatch(&exh->latch);
tDebug("msg refId: %" PRId64 "", handle);
(void)transReleaseExHandle(transGetRefMgt(), handle);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return 0;
} else {
exh->inited = 1;
taosWUnLockLatch(&exh->latch);
(void)transReleaseExHandle(transGetRefMgt(), handle);
TAOS_UNUSED(transReleaseExHandle(transGetRefMgt(), handle));
}
}
}
@ -2976,16 +2973,16 @@ int32_t transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, S
EPSET_GET_INUSE_IP(pEpSet), EPSET_GET_INUSE_PORT(pEpSet), pReq->info.ahandle);
if ((code = transAsyncSend(pThrd->asyncPool, &(pCliMsg->q))) != 0) {
destroyCmsg(pCliMsg);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return (code == TSDB_CODE_RPC_ASYNC_MODULE_QUIT ? TSDB_CODE_RPC_MODULE_QUIT : code);
}
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return 0;
_exception:
transFreeMsg(pReq->pCont);
pReq->pCont = NULL;
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return code;
}
int32_t transSendRequestWithId(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, int64_t* transpointId) {
@ -3021,16 +3018,16 @@ int32_t transSendRequestWithId(void* shandle, const SEpSet* pEpSet, STransMsg* p
EPSET_GET_INUSE_IP(pEpSet), EPSET_GET_INUSE_PORT(pEpSet), pReq->info.ahandle);
if ((code = transAsyncSend(pThrd->asyncPool, &(pCliMsg->q))) != 0) {
destroyCmsg(pCliMsg);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return (code == TSDB_CODE_RPC_ASYNC_MODULE_QUIT ? TSDB_CODE_RPC_MODULE_QUIT : code);
}
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return 0;
_exception:
transFreeMsg(pReq->pCont);
pReq->pCont = NULL;
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return code;
}
@ -3068,7 +3065,7 @@ int32_t transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx));
if (pCtx == NULL) {
(void)tsem_destroy(sem);
TAOS_UNUSED(tsem_destroy(sem));
taosMemoryFree(sem);
TAOS_CHECK_GOTO(terrno, NULL, _RETURN1);
}
@ -3082,7 +3079,7 @@ int32_t transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg));
if (cliMsg == NULL) {
(void)tsem_destroy(sem);
TAOS_UNUSED(tsem_destroy(sem));
taosMemoryFree(sem);
taosMemoryFree(pCtx);
TAOS_CHECK_GOTO(terrno, NULL, _RETURN1);
@ -3103,18 +3100,18 @@ int32_t transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
destroyCmsg(cliMsg);
TAOS_CHECK_GOTO((code == TSDB_CODE_RPC_ASYNC_MODULE_QUIT ? TSDB_CODE_RPC_MODULE_QUIT : code), NULL, _RETURN);
}
(void)tsem_wait(sem);
TAOS_UNUSED(tsem_wait(sem));
memcpy(pRsp, pTransRsp, sizeof(STransMsg));
_RETURN:
tsem_destroy(sem);
taosMemoryFree(sem);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
taosMemoryFree(pTransRsp);
return code;
_RETURN1:
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
taosMemoryFree(pTransRsp);
taosMemoryFree(pReq->pCont);
pReq->pCont = NULL;
@ -3151,7 +3148,7 @@ int32_t transCreateSyncMsg(STransMsg* pTransMsg, int64_t* refId) {
return 0;
_EXIT:
(void)tsem2_destroy(sem);
TAOS_UNUSED(tsem2_destroy(sem));
taosMemoryFree(sem);
taosMemoryFree(pSyncMsg);
return code;
@ -3235,15 +3232,15 @@ int32_t transSendRecvWithTimeout(void* shandle, SEpSet* pEpSet, STransMsg* pReq,
}
}
_RETURN:
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
(void)taosReleaseRef(transGetSyncMsgMgt(), ref);
(void)taosRemoveRef(transGetSyncMsgMgt(), ref);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
TAOS_UNUSED(taosReleaseRef(transGetSyncMsgMgt(), ref));
TAOS_UNUSED(taosRemoveRef(transGetSyncMsgMgt(), ref));
return code;
_RETURN2:
transFreeMsg(pReq->pCont);
pReq->pCont = NULL;
taosMemoryFree(pTransMsg);
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return code;
}
/*
@ -3295,7 +3292,7 @@ int32_t transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn) {
}
}
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)shandle));
return code;
}

24
source/libs/transport/src/transComm.c
View File

@ -272,7 +272,7 @@ int32_t transAsyncPoolCreate(uv_loop_t* loop, int sz, void* arg, AsyncCB cb, SAs
}
item->pThrd = arg;
QUEUE_INIT(&item->qmsg);
(void)taosThreadMutexInit(&item->mtx, NULL);
TAOS_UNUSED(taosThreadMutexInit(&item->mtx, NULL));
async->data = item;
err = uv_async_init(loop, async, cb);
@ -301,7 +301,7 @@ void transAsyncPoolDestroy(SAsyncPool* pool) {
SAsyncItem* item = async->data;
if (item == NULL) continue;
(void)taosThreadMutexDestroy(&item->mtx);
TAOS_UNUSED(taosThreadMutexDestroy(&item->mtx));
taosMemoryFree(item);
}
taosMemoryFree(pool->asyncs);
@ -328,9 +328,9 @@ int transAsyncSend(SAsyncPool* pool, queue* q) {
uv_async_t* async = &(pool->asyncs[idx]);
SAsyncItem* item = async->data;
(void)taosThreadMutexLock(&item->mtx);
TAOS_UNUSED(taosThreadMutexLock(&item->mtx));
QUEUE_PUSH(&item->qmsg, q);
(void)taosThreadMutexUnlock(&item->mtx);
TAOS_UNUSED(taosThreadMutexUnlock(&item->mtx));
int ret = uv_async_send(async);
if (ret != 0) {
tError("failed to send async,reason:%s", uv_err_name(ret));
@ -393,7 +393,7 @@ void* transCtxDumpVal(STransCtx* ctx, int32_t key) {
return NULL;
}
void* ret = NULL;
(void)(*cVal->clone)(cVal->val, &ret);
TAOS_UNUSED((*cVal->clone)(cVal->val, &ret));
return ret;
}
void* transCtxDumpBrokenlinkVal(STransCtx* ctx, int32_t* msgType) {
@ -401,7 +401,7 @@ void* transCtxDumpBrokenlinkVal(STransCtx* ctx, int32_t* msgType) {
if (ctx->brokenVal.clone == NULL) {
return ret;
}
(void)(*ctx->brokenVal.clone)(ctx->brokenVal.val, &ret);
TAOS_UNUSED((*ctx->brokenVal.clone)(ctx->brokenVal.val, &ret));
*msgType = ctx->brokenVal.msgType;
@ -549,7 +549,7 @@ static void transDQTimeout(uv_timer_t* timer) {
}
} while (1);
if (timeout != 0) {
(void)uv_timer_start(queue->timer, transDQTimeout, timeout, 0);
TAOS_UNUSED(uv_timer_start(queue->timer, transDQTimeout, timeout, 0));
}
}
int32_t transDQCreate(uv_loop_t* loop, SDelayQueue** queue) {
@ -614,7 +614,7 @@ void transDQDestroy(SDelayQueue* queue, void (*freeFunc)(void* arg)) {
taosMemoryFree(queue);
}
void transDQCancel(SDelayQueue* queue, SDelayTask* task) {
(void)uv_timer_stop(queue->timer);
TAOS_UNUSED(uv_timer_stop(queue->timer));
if (heapSize(queue->heap) <= 0) {
taosMemoryFree(task->arg);
@ -634,7 +634,7 @@ void transDQCancel(SDelayQueue* queue, SDelayTask* task) {
SDelayTask* task = container_of(minNode, SDelayTask, node);
uint64_t timeout = now > task->execTime ? now - task->execTime : 0;
(void)uv_timer_start(queue->timer, transDQTimeout, timeout, 0);
TAOS_UNUSED(uv_timer_start(queue->timer, transDQTimeout, timeout, 0));
}
}
@ -659,7 +659,7 @@ SDelayTask* transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg,
tTrace("timer %p put task into delay queue, timeoutMs:%" PRIu64, queue->timer, timeoutMs);
heapInsert(queue->heap, &task->node);
(void)uv_timer_start(queue->timer, transDQTimeout, timeoutMs, 0);
TAOS_UNUSED(uv_timer_start(queue->timer, transDQTimeout, timeoutMs, 0));
return task;
}
@ -708,7 +708,7 @@ static void transInitEnv() {
svrRefMgt = transOpenRefMgt(50000, transDestroyExHandle);
instMgt = taosOpenRef(50, rpcCloseImpl);
transSyncMsgMgt = taosOpenRef(50, transDestroySyncMsg);
(void)uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1");
TAOS_UNUSED(uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1"));
}
static void transDestroyEnv() {
transCloseRefMgt(refMgt);
@ -775,7 +775,7 @@ void transDestroySyncMsg(void* msg) {
if (msg == NULL) return;
STransSyncMsg* pSyncMsg = msg;
(void)tsem2_destroy(pSyncMsg->pSem);
TAOS_UNUSED(tsem2_destroy(pSyncMsg->pSem));
taosMemoryFree(pSyncMsg->pSem);
transFreeMsg(pSyncMsg->pRsp->pCont);
taosMemoryFree(pSyncMsg->pRsp);

85
source/libs/transport/src/transSvr.c
View File

@ -188,11 +188,11 @@ static void sendQuitToWorkThrd(SWorkThrd* pThrd);
static int32_t addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName);
static int32_t addHandleToAcceptloop(void* arg);
#define SRV_RELEASE_UV(loop) \
do { \
(void)uv_walk(loop, uvWalkCb, NULL); \
(void)uv_run(loop, UV_RUN_DEFAULT); \
(void)uv_loop_close(loop); \
#define SRV_RELEASE_UV(loop) \
do { \
TAOS_UNUSED(uv_walk(loop, uvWalkCb, NULL)); \
TAOS_UNUSED(uv_run(loop, UV_RUN_DEFAULT)); \
TAOS_UNUSED(uv_loop_close(loop)); \
} while (0);
#define ASYNC_ERR_JRET(thrd) \
@ -594,7 +594,7 @@ void uvOnSendCb(uv_write_t* req, int status) {
(pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL);
memset(&conn->regArg, 0, sizeof(conn->regArg));
}
(void)transQueuePop(&conn->srvMsgs);
TAOS_UNUSED(transQueuePop(&conn->srvMsgs));
taosMemoryFree(msg);
msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0);
@ -651,7 +651,7 @@ static int uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) {
// handle invalid drop_task resp, TD-20098
if (pConn->inType == TDMT_SCH_DROP_TASK && pMsg->code == TSDB_CODE_VND_INVALID_VGROUP_ID) {
(void)transQueuePop(&pConn->srvMsgs);
TAOS_UNUSED(transQueuePop(&pConn->srvMsgs));
destroySmsg(smsg);
return TSDB_CODE_INVALID_MSG;
}
@ -707,7 +707,7 @@ static FORCE_INLINE void uvStartSendRespImpl(SSvrMsg* smsg) {
transRefSrvHandle(pConn);
uv_write_t* req = transReqQueuePush(&pConn->wreqQueue);
(void)uv_write(req, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnSendCb);
TAOS_UNUSED(uv_write(req, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnSendCb));
}
static void uvStartSendResp(SSvrMsg* smsg) {
// impl
@ -759,9 +759,9 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
queue wq;
// batch process to avoid to lock/unlock frequently
(void)taosThreadMutexLock(&item->mtx);
TAOS_UNUSED(taosThreadMutexLock(&item->mtx));
QUEUE_MOVE(&item->qmsg, &wq);
(void)taosThreadMutexUnlock(&item->mtx);
TAOS_UNUSED(taosThreadMutexUnlock(&item->mtx));
while (!QUEUE_IS_EMPTY(&wq)) {
queue* head = QUEUE_HEAD(&wq);
@ -784,12 +784,12 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
SExHandle* exh2 = transAcquireExHandle(transGetSvrRefMgt(), refId);
if (exh2 == NULL || exh1 != exh2) {
tTrace("handle except msg %p, ignore it", exh1);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
destroySmsg(msg);
continue;
}
msg->pConn = exh1->handle;
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
(*transAsyncHandle[msg->type])(msg, pThrd);
}
}
@ -827,7 +827,7 @@ static bool uvRecvReleaseReq(SSvrConn* pConn, STransMsgHead* pHead) {
tTrace("conn %p received release request", pConn);
STraceId traceId = pHead->traceId;
(void)transClearBuffer(&pConn->readBuf);
TAOS_UNUSED(transClearBuffer(&pConn->readBuf));
transFreeMsg(transContFromHead((char*)pHead));
if (pConn->status != ConnAcquire) {
return true;
@ -906,7 +906,8 @@ void uvOnAcceptCb(uv_stream_t* stream, int status) {
tTrace("new connection accepted by main server, dispatch to %dth worker-thread", pObj->workerIdx);
(void)uv_write2(wr, (uv_stream_t*)&(pObj->pipe[pObj->workerIdx][0]), &buf, 1, (uv_stream_t*)cli, uvOnPipeWriteCb);
TAOS_UNUSED(
uv_write2(wr, (uv_stream_t*)&(pObj->pipe[pObj->workerIdx][0]), &buf, 1, (uv_stream_t*)cli, uvOnPipeWriteCb));
} else {
if (!uv_is_closing((uv_handle_t*)cli)) {
tError("failed to accept tcp: %s", uv_err_name(err));
@ -971,7 +972,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
if (uv_accept(q, (uv_stream_t*)(pConn->pTcp)) == 0) {
uv_os_fd_t fd;
(void)uv_fileno((const uv_handle_t*)pConn->pTcp, &fd);
TAOS_UNUSED(uv_fileno((const uv_handle_t*)pConn->pTcp, &fd));
tTrace("conn %p created, fd:%d", pConn, fd);
struct sockaddr peername, sockname;
@ -981,7 +982,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
transUnrefSrvHandle(pConn);
return;
}
(void)transSockInfo2Str(&peername, pConn->dst);
TAOS_UNUSED(transSockInfo2Str(&peername, pConn->dst));
addrlen = sizeof(sockname);
if (0 != uv_tcp_getsockname(pConn->pTcp, (struct sockaddr*)&sockname, &addrlen)) {
@ -989,7 +990,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
transUnrefSrvHandle(pConn);
return;
}
(void)transSockInfo2Str(&sockname, pConn->src);
TAOS_UNUSED(transSockInfo2Str(&sockname, pConn->src));
struct sockaddr_in addr = *(struct sockaddr_in*)&peername;
struct sockaddr_in saddr = *(struct sockaddr_in*)&sockname;
@ -998,7 +999,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
pConn->serverIp = saddr.sin_addr.s_addr;
pConn->port = ntohs(addr.sin_port);
(void)uv_read_start((uv_stream_t*)(pConn->pTcp), uvAllocRecvBufferCb, uvOnRecvCb);
TAOS_UNUSED(uv_read_start((uv_stream_t*)(pConn->pTcp), uvAllocRecvBufferCb, uvOnRecvCb));
} else {
tDebug("failed to create new connection");
@ -1010,7 +1011,7 @@ void* transAcceptThread(void* arg) {
// opt
setThreadName("trans-accept");
SServerObj* srv = (SServerObj*)arg;
(void)uv_run(srv->loop, UV_RUN_DEFAULT);
TAOS_UNUSED(uv_run(srv->loop, UV_RUN_DEFAULT));
return NULL;
}
@ -1021,7 +1022,7 @@ void uvOnPipeConnectionCb(uv_connect_t* connect, int status) {
};
SWorkThrd* pThrd = container_of(connect, SWorkThrd, connect_req);
(void)uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
TAOS_UNUSED(uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb));
}
static int32_t addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) {
int32_t code = 0;
@ -1124,7 +1125,7 @@ static int32_t addHandleToAcceptloop(void* arg) {
void* transWorkerThread(void* arg) {
setThreadName("trans-svr-work");
SWorkThrd* pThrd = (SWorkThrd*)arg;
(void)uv_run(pThrd->loop, UV_RUN_DEFAULT);
TAOS_UNUSED(uv_run(pThrd->loop, UV_RUN_DEFAULT));
return NULL;
}
@ -1201,7 +1202,7 @@ static FORCE_INLINE SSvrConn* createConn(void* hThrd) {
_end:
if (pConn) {
transQueueDestroy(&pConn->srvMsgs);
(void)transDestroyBuffer(&pConn->readBuf);
TAOS_UNUSED(transDestroyBuffer(&pConn->readBuf));
taosMemoryFree(pConn->pTcp);
taosMemoryFree(pConn);
pConn = NULL;
@ -1230,8 +1231,8 @@ static FORCE_INLINE void destroyConnRegArg(SSvrConn* conn) {
}
static int32_t reallocConnRef(SSvrConn* conn) {
if (conn->refId > 0) {
(void)transReleaseExHandle(transGetSvrRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetSvrRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetSvrRefMgt(), conn->refId));
}
// avoid app continue to send msg on invalid handle
SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
@ -1267,8 +1268,8 @@ static void uvDestroyConn(uv_handle_t* handle) {
}
SWorkThrd* thrd = conn->hostThrd;
(void)transReleaseExHandle(transGetSvrRefMgt(), conn->refId);
(void)transRemoveExHandle(transGetSvrRefMgt(), conn->refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), conn->refId));
TAOS_UNUSED(transRemoveExHandle(transGetSvrRefMgt(), conn->refId));
STrans* pTransInst = thrd->pTransInst;
tDebug("%s conn %p destroy", transLabel(pTransInst), conn);
@ -1283,7 +1284,7 @@ static void uvDestroyConn(uv_handle_t* handle) {
QUEUE_REMOVE(&conn->queue);
taosMemoryFree(conn->pTcp);
destroyConnRegArg(conn);
(void)transDestroyBuffer(&conn->readBuf);
TAOS_UNUSED(transDestroyBuffer(&conn->readBuf));
taosMemoryFree(conn);
if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) {
@ -1563,7 +1564,7 @@ void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd) {
if (!transQueuePush(&conn->srvMsgs, msg)) {
return;
}
(void)transQueuePop(&conn->srvMsgs);
TAOS_UNUSED(transQueuePop(&conn->srvMsgs));
if (conn->regArg.init) {
transFreeMsg(conn->regArg.msg.pCont);
@ -1636,7 +1637,7 @@ void destroyWorkThrd(SWorkThrd* pThrd) {
}
if (pThrd->inited) {
sendQuitToWorkThrd(pThrd);
(void)taosThreadJoin(pThrd->thread, NULL);
TAOS_UNUSED(taosThreadJoin(pThrd->thread, NULL));
SRV_RELEASE_UV(pThrd->loop);
TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSvrMsg, destroySmsgWrapper, NULL);
}
@ -1646,7 +1647,7 @@ void sendQuitToWorkThrd(SWorkThrd* pThrd) {
SSvrMsg* msg = taosMemoryCalloc(1, sizeof(SSvrMsg));
msg->type = Quit;
tDebug("server send quit msg to work thread");
(void)transAsyncSend(pThrd->asyncPool, &msg->q);
TAOS_UNUSED(transAsyncSend(pThrd->asyncPool, &msg->q));
}
void transCloseServer(void* arg) {
@ -1655,8 +1656,8 @@ void transCloseServer(void* arg) {
if (srv->inited) {
tDebug("send quit msg to accept thread");
(void)uv_async_send(srv->pAcceptAsync);
(void)taosThreadJoin(srv->thread, NULL);
TAOS_UNUSED(uv_async_send(srv->pAcceptAsync));
TAOS_UNUSED(taosThreadJoin(srv->thread, NULL));
SRV_RELEASE_UV(srv->loop);
for (int i = 0; i < srv->numOfThreads; i++) {
@ -1724,15 +1725,15 @@ int32_t transReleaseSrvHandle(void* handle) {
tDebug("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle);
if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) {
destroySmsg(m);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
}
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return 0;
_return1:
tDebug("handle %p failed to send to release handle", exh);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
_return2:
tDebug("handle %p failed to send to release handle", exh);
@ -1775,17 +1776,17 @@ int32_t transSendResponse(const STransMsg* msg) {
tGDebug("conn %p start to send resp (1/2)", exh->handle);
if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) {
destroySmsg(m);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
}
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return 0;
_return1:
tDebug("handle %p failed to send resp", exh);
rpcFreeCont(msg->pCont);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
_return2:
tDebug("handle %p failed to send resp", exh);
@ -1820,17 +1821,17 @@ int32_t transRegisterMsg(const STransMsg* msg) {
tDebug("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle);
if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) {
destroySmsg(m);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
}
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return 0;
_return1:
tDebug("handle %p failed to register brokenlink", exh);
rpcFreeCont(msg->pCont);
(void)transReleaseExHandle(transGetSvrRefMgt(), refId);
TAOS_UNUSED(transReleaseExHandle(transGetSvrRefMgt(), refId));
return code;
_return2:
tDebug("handle %p failed to register brokenlink", exh);
@ -1875,7 +1876,7 @@ int32_t transSetIpWhiteList(void* thandle, void* arg, FilteFunc* func) {
break;
}
}
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)thandle);
TAOS_UNUSED(transReleaseExHandle(transGetInstMgt(), (int64_t)thandle));
if (code != 0) {
tError("ip-white-list update failed since %s", tstrerror(code));