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

Merge pull request #26869 from taosdata/fix/fixErrorRetRocksdb 

refactor erro ret
This commit is contained in:
Hongze Cheng 2024-07-30 16:07:19 +08:00 committed by GitHub
parent 2a5bb9d06d 417392ea23
commit 623bd144d6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 20 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c
View File

@ -214,8 +214,17 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
void dmSendNotifyReq(SDnodeMgmt *pMgmt, SNotifyReq *pReq) { void dmSendNotifyReq(SDnodeMgmt *pMgmt, SNotifyReq *pReq) {
int32_t contLen = tSerializeSNotifyReq(NULL, 0, pReq); int32_t contLen = tSerializeSNotifyReq(NULL, 0, pReq);
void *pHead = rpcMallocCont(contLen); if (contLen < 0) {
tSerializeSNotifyReq(pHead, contLen, pReq); dError("failed to serialize notify req since %s", tstrerror(contLen));
return;
}
void *pHead = rpcMallocCont(contLen);
contLen = tSerializeSNotifyReq(pHead, contLen, pReq);
if (contLen < 0) {
rpcFreeCont(pHead);
dError("failed to serialize notify req since %s", tstrerror(contLen));
return;
}
SRpcMsg rpcMsg = {.pCont = pHead, SRpcMsg rpcMsg = {.pCont = pHead,
.contLen = contLen, .contLen = contLen,

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

@ -1126,7 +1126,7 @@ int32_t chkpMayDelObsolete(void* arg, int64_t chkpId, char* path) {
taosArrayDestroy(pBackend->chkpSaved); taosArrayDestroy(pBackend->chkpSaved);
pBackend->chkpSaved = chkpDup; pBackend->chkpSaved = chkpDup;
taosThreadRwlockUnlock(&pBackend->chkpDirLock); (void)taosThreadRwlockUnlock(&pBackend->chkpDirLock);
for (int i = 0; i < taosArrayGetSize(chkpDel); i++) { for (int i = 0; i < taosArrayGetSize(chkpDel); i++) {
int64_t id = *(int64_t*)taosArrayGet(chkpDel, i); int64_t id = *(int64_t*)taosArrayGet(chkpDel, i);
@ -1438,7 +1438,7 @@ int32_t taskDbBuildSnap(void* arg, SArray* pSnap) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
break; break;
} }
taosArrayPush(pSnap, &snap); (void)taosArrayPush(pSnap, &snap);
pIter = taosHashIterate(pMeta->pTaskDbUnique, pIter); pIter = taosHashIterate(pMeta->pTaskDbUnique, pIter);
} }
@ -1505,7 +1505,7 @@ void* taskAcquireDb(int64_t refId) {
} }
void taskReleaseDb(int64_t refId) { void taskReleaseDb(int64_t refId) {
// release // release
taosReleaseRef(taskDbWrapperId, refId); (void)taosReleaseRef(taskDbWrapperId, refId);
} }
int64_t taskGetDBRef(void* arg) { int64_t taskGetDBRef(void* arg) {
@ -1566,7 +1566,7 @@ int32_t chkpLoadExtraInfo(char* pChkpIdDir, int64_t* chkpId, int64_t* processId)
code = 0; code = 0;
_EXIT: _EXIT:
taosMemoryFree(pDst); taosMemoryFree(pDst);
taosCloseFile(&pFile); (void)taosCloseFile(&pFile);
return code; return code;
} }
int32_t chkpAddExtraInfo(char* pChkpIdDir, int64_t chkpId, int64_t processId) { int32_t chkpAddExtraInfo(char* pChkpIdDir, int64_t chkpId, int64_t processId) {
@ -4094,7 +4094,7 @@ void streamStateSessionClear_rocksdb(SStreamState* pState) {
if (code == 0 && size > 0) { if (code == 0 && size > 0) {
memset(buf, 0, size); memset(buf, 0, size);
// refactor later // refactor later
streamStateSessionPut_rocksdb(pState, &delKey, buf, size); (void)streamStateSessionPut_rocksdb(pState, &delKey, buf, size);
} else { } else {
taosMemoryFreeClear(buf); taosMemoryFreeClear(buf);
break; break;
@ -4304,7 +4304,7 @@ void streamStateDestroyBatch(void* pBatch) { rocksdb_writebatch_destroy((rock
int32_t streamStatePutBatch(SStreamState* pState, const char* cfKeyName, rocksdb_writebatch_t* pBatch, void* key, int32_t streamStatePutBatch(SStreamState* pState, const char* cfKeyName, rocksdb_writebatch_t* pBatch, void* key,
void* val, int32_t vlen, int64_t ttl) { void* val, int32_t vlen, int64_t ttl) {
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend; STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
atomic_add_fetch_64(&wrapper->dataWritten, 1); (void)atomic_add_fetch_64(&wrapper->dataWritten, 1);
int i = streamStateGetCfIdx(pState, cfKeyName); int i = streamStateGetCfIdx(pState, cfKeyName);
if (i < 0) { if (i < 0) {
@ -4350,7 +4350,7 @@ int32_t streamStatePutBatchOptimize(SStreamState* pState, int32_t cfIdx, rocksdb
{ {
char tbuf[256] = {0}; char tbuf[256] = {0};
ginitDict[cfIdx].toStrFunc((void*)key, tbuf); (void)(ginitDict[cfIdx].toStrFunc((void*)key, tbuf));
stTrace("streamState str: %s succ to write to %s_%s", tbuf, wrapper->idstr, ginitDict[cfIdx].key); stTrace("streamState str: %s succ to write to %s_%s", tbuf, wrapper->idstr, ginitDict[cfIdx].key);
} }
return 0; return 0;
@ -4358,7 +4358,7 @@ int32_t streamStatePutBatchOptimize(SStreamState* pState, int32_t cfIdx, rocksdb
int32_t streamStatePutBatch_rocksdb(SStreamState* pState, void* pBatch) { int32_t streamStatePutBatch_rocksdb(SStreamState* pState, void* pBatch) {
char* err = NULL; char* err = NULL;
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend; STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
atomic_add_fetch_64(&wrapper->dataWritten, 1); (void)atomic_add_fetch_64(&wrapper->dataWritten, 1);
rocksdb_write(wrapper->db, wrapper->writeOpt, (rocksdb_writebatch_t*)pBatch, &err); rocksdb_write(wrapper->db, wrapper->writeOpt, (rocksdb_writebatch_t*)pBatch, &err);
if (err != NULL) { if (err != NULL) {
stError("streamState failed to write batch, err:%s", err); stError("streamState failed to write batch, err:%s", err);
@ -4738,7 +4738,7 @@ int32_t dbChkpDumpTo(SDbChkp* p, char* dname, SArray* list) {
static char* chkpMeta = "META"; static char* chkpMeta = "META";
int32_t code = 0; int32_t code = 0;
taosThreadRwlockRdlock(&p->rwLock); (void)taosThreadRwlockRdlock(&p->rwLock);
int32_t cap = p->len + 128; int32_t cap = p->len + 128;