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

Merge pull request #29472 from taosdata/fix/fixInvalidLock 

fix invalid lock
This commit is contained in:
Hongze Cheng 2025-01-06 18:55:55 +08:00 committed by GitHub
parent 537d110265 c372b5da51
commit 2c2654c10d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 18 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
source/libs/stream/src/streamMeta.c
View File

@ -253,7 +253,7 @@ _EXIT:
int32_t len = strlen(pMeta->path) + 32; int32_t len = strlen(pMeta->path) + 32;
char* state = taosMemoryCalloc(1, len); char* state = taosMemoryCalloc(1, len);
if (state != NULL) { if (state != NULL) {
(void) snprintf(state, len, "%s%s%s", pMeta->path, TD_DIRSEP, "state"); (void)snprintf(state, len, "%s%s%s", pMeta->path, TD_DIRSEP, "state");
taosRemoveDir(state); taosRemoveDir(state);
taosMemoryFree(state); taosMemoryFree(state);
} else { } else {
@ -380,7 +380,7 @@ int32_t streamMetaOpen(const char* path, void* ahandle, FTaskBuild buildTaskFn,
char* tpath = taosMemoryCalloc(1, len); char* tpath = taosMemoryCalloc(1, len);
TSDB_CHECK_NULL(tpath, code, lino, _err, terrno); TSDB_CHECK_NULL(tpath, code, lino, _err, terrno);
(void) snprintf(tpath, len, "%s%s%s", path, TD_DIRSEP, "stream"); (void)snprintf(tpath, len, "%s%s%s", path, TD_DIRSEP, "stream");
pMeta->path = tpath; pMeta->path = tpath;
code = streamMetaOpenTdb(pMeta); code = streamMetaOpenTdb(pMeta);
@ -392,6 +392,22 @@ int32_t streamMetaOpen(const char* path, void* ahandle, FTaskBuild buildTaskFn,
TSDB_CHECK_CODE(code, lino, _err); TSDB_CHECK_CODE(code, lino, _err);
} }
// set the attribute when running on Linux OS
TdThreadRwlockAttr attr;
code = taosThreadRwlockAttrInit(&attr);
TSDB_CHECK_CODE(code, lino, _err);
#ifdef LINUX
code = pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
TSDB_CHECK_CODE(code, lino, _err);
#endif
code = taosThreadRwlockInit(&pMeta->lock, &attr);
TSDB_CHECK_CODE(code, lino, _err);
code = taosThreadRwlockAttrDestroy(&attr);
TSDB_CHECK_CODE(code, lino, _err);
if ((code = streamMetaBegin(pMeta) < 0)) { if ((code = streamMetaBegin(pMeta) < 0)) {
stError("vgId:%d begin trans for stream meta failed", pMeta->vgId); stError("vgId:%d begin trans for stream meta failed", pMeta->vgId);
goto _err; goto _err;
@ -431,22 +447,6 @@ int32_t streamMetaOpen(const char* path, void* ahandle, FTaskBuild buildTaskFn,
stInfo("vgId:%d open stream meta succ, latest checkpoint:%" PRId64 ", stage:%" PRId64, vgId, pMeta->chkpId, stage); stInfo("vgId:%d open stream meta succ, latest checkpoint:%" PRId64 ", stage:%" PRId64, vgId, pMeta->chkpId, stage);
// set the attribute when running on Linux OS
TdThreadRwlockAttr attr;
code = taosThreadRwlockAttrInit(&attr);
TSDB_CHECK_CODE(code, lino, _err);
#ifdef LINUX
code = pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
TSDB_CHECK_CODE(code, lino, _err);
#endif
code = taosThreadRwlockInit(&pMeta->lock, &attr);
TSDB_CHECK_CODE(code, lino, _err);
code = taosThreadRwlockAttrDestroy(&attr);
TSDB_CHECK_CODE(code, lino, _err);
code = bkdMgtCreate(tpath, (SBkdMgt**)&pMeta->bkdChkptMgt); code = bkdMgtCreate(tpath, (SBkdMgt**)&pMeta->bkdChkptMgt);
TSDB_CHECK_CODE(code, lino, _err); TSDB_CHECK_CODE(code, lino, _err);