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

refactor: changel lockfree to rwlock

This commit is contained in:
Shengliang Guan 2022-05-19 15:08:32 +08:00
parent 568c6269dc
commit c5d18b5afd
4 changed files with 29 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
View File

@ -26,21 +26,21 @@ extern "C" {
#endif #endif
typedef struct SVnodeMgmt { typedef struct SVnodeMgmt {
SDnodeData *pData; SDnodeData *pData;
SMsgCb msgCb; SMsgCb msgCb;
const char *path; const char *path;
const char *name; const char *name;
SQWorkerPool queryPool; SQWorkerPool queryPool;
SQWorkerPool fetchPool; SQWorkerPool fetchPool;
SWWorkerPool syncPool; SWWorkerPool syncPool;
SWWorkerPool writePool; SWWorkerPool writePool;
SWWorkerPool mergePool; SWWorkerPool mergePool;
SSingleWorker mgmtWorker; SSingleWorker mgmtWorker;
SSingleWorker monitorWorker; SSingleWorker monitorWorker;
SHashObj *hash; SHashObj *hash;
SRWLatch latch; TdThreadRwlock lock;
SVnodesStat state; SVnodesStat state;
STfs *pTfs; STfs *pTfs;
} SVnodeMgmt; } SVnodeMgmt;
typedef struct { typedef struct {

4
source/dnode/mgmt/mgmt_vnode/src/vmFile.c
View File

@ -17,7 +17,7 @@
#include "vmInt.h" #include "vmInt.h"
SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) { SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) {
taosRLockLatch(&pMgmt->latch); taosThreadRwlockRdlock(&pMgmt->lock);
int32_t num = 0; int32_t num = 0;
int32_t size = taosHashGetSize(pMgmt->hash); int32_t size = taosHashGetSize(pMgmt->hash);
@ -38,7 +38,7 @@ SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) {
} }
} }
taosRUnLockLatch(&pMgmt->latch); taosThreadRwlockUnlock(&pMgmt->lock);
*numOfVnodes = num; *numOfVnodes = num;
return pVnodes; return pVnodes;

4
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
View File

@ -20,7 +20,7 @@ void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
pInfo->pVloads = taosArrayInit(pMgmt->state.totalVnodes, sizeof(SVnodeLoad)); pInfo->pVloads = taosArrayInit(pMgmt->state.totalVnodes, sizeof(SVnodeLoad));
if (pInfo->pVloads == NULL) return; if (pInfo->pVloads == NULL) return;
taosRLockLatch(&pMgmt->latch); taosThreadRwlockRdlock(&pMgmt->lock);
void *pIter = taosHashIterate(pMgmt->hash, NULL); void *pIter = taosHashIterate(pMgmt->hash, NULL);
while (pIter) { while (pIter) {
@ -34,7 +34,7 @@ void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
pIter = taosHashIterate(pMgmt->hash, pIter); pIter = taosHashIterate(pMgmt->hash, pIter);
} }
taosRUnLockLatch(&pMgmt->latch); taosThreadRwlockUnlock(&pMgmt->lock);
} }
void vmGetMonitorInfo(SVnodeMgmt *pMgmt, SMonVmInfo *pInfo) { void vmGetMonitorInfo(SVnodeMgmt *pMgmt, SMonVmInfo *pInfo) {

19
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
View File

@ -20,14 +20,14 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
SVnodeObj *pVnode = NULL; SVnodeObj *pVnode = NULL;
int32_t refCount = 0; int32_t refCount = 0;
taosRLockLatch(&pMgmt->latch); taosThreadRwlockRdlock(&pMgmt->lock);
taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode); taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode);
if (pVnode == NULL) { if (pVnode == NULL) {
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID; terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
} else { } else {
refCount = atomic_add_fetch_32(&pVnode->refCount, 1); refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
} }
taosRUnLockLatch(&pMgmt->latch); taosThreadRwlockUnlock(&pMgmt->lock);
if (pVnode != NULL) { if (pVnode != NULL) {
dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount); dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount);
@ -39,9 +39,9 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
if (pVnode == NULL) return; if (pVnode == NULL) return;
taosRLockLatch(&pMgmt->latch); taosThreadRwlockRdlock(&pMgmt->lock);
int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1); int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1);
taosRUnLockLatch(&pMgmt->latch); taosThreadRwlockUnlock(&pMgmt->lock);
dTrace("vgId:%d, release vnode, refCount:%d", pVnode->vgId, refCount); dTrace("vgId:%d, release vnode, refCount:%d", pVnode->vgId, refCount);
} }
@ -70,9 +70,9 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
return -1; return -1;
} }
taosWLockLatch(&pMgmt->latch); taosThreadRwlockWrlock(&pMgmt->lock);
int32_t code = taosHashPut(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *)); int32_t code = taosHashPut(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *));
taosWUnLockLatch(&pMgmt->latch); taosThreadRwlockUnlock(&pMgmt->lock);
return code; return code;
} }
@ -80,9 +80,9 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
char path[TSDB_FILENAME_LEN] = {0}; char path[TSDB_FILENAME_LEN] = {0};
taosWLockLatch(&pMgmt->latch); taosThreadRwlockWrlock(&pMgmt->lock);
taosHashRemove(pMgmt->hash, &pVnode->vgId, sizeof(int32_t)); taosHashRemove(pMgmt->hash, &pVnode->vgId, sizeof(int32_t));
taosWUnLockLatch(&pMgmt->latch); taosThreadRwlockUnlock(&pMgmt->lock);
vmReleaseVnode(pMgmt, pVnode); vmReleaseVnode(pMgmt, pVnode);
while (pVnode->refCount > 0) taosMsleep(10); while (pVnode->refCount > 0) taosMsleep(10);
@ -239,6 +239,7 @@ static void vmCleanup(SVnodeMgmt *pMgmt) {
vmStopWorker(pMgmt); vmStopWorker(pMgmt);
vnodeCleanup(); vnodeCleanup();
tfsClose(pMgmt->pTfs); tfsClose(pMgmt->pTfs);
taosThreadRwlockDestroy(&pMgmt->lock);
taosMemoryFree(pMgmt); taosMemoryFree(pMgmt);
} }
@ -260,7 +261,7 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
pMgmt->msgCb.queueFps[MERGE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToMergeQueue; pMgmt->msgCb.queueFps[MERGE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToMergeQueue;
pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)vmGetQueueSize; pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)vmGetQueueSize;
pMgmt->msgCb.mgmt = pMgmt; pMgmt->msgCb.mgmt = pMgmt;
taosInitRWLatch(&pMgmt->latch); taosThreadRwlockInit(&pMgmt->lock, NULL);
SDiskCfg dCfg = {0}; SDiskCfg dCfg = {0};
tstrncpy(dCfg.dir, tsDataDir, TSDB_FILENAME_LEN); tstrncpy(dCfg.dir, tsDataDir, TSDB_FILENAME_LEN);