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

fix thread conflict

This commit is contained in:
Shengliang Guan 2022-03-20 10:19:14 +08:00
parent 3544df4112
commit 4cb813a920
10 changed files with 52 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
include/dnode/bnode/bnode.h
View File

@ -71,13 +71,6 @@ int32_t bndGetLoad(SBnode *pBnode, SBnodeLoad *pLoad);
*/ */
int32_t bndProcessWMsgs(SBnode *pBnode, SArray *pMsgs); int32_t bndProcessWMsgs(SBnode *pBnode, SArray *pMsgs);
/**
* @brief Drop a bnode.
*
* @param path Path of the bnode.
*/
void bndDestroy(const char *path);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

2
source/dnode/bnode/src/bnode.c
View File

@ -25,5 +25,3 @@ void bndClose(SBnode *pBnode) { free(pBnode); }
int32_t bndGetLoad(SBnode *pBnode, SBnodeLoad *pLoad) { return 0; } int32_t bndGetLoad(SBnode *pBnode, SBnodeLoad *pLoad) { return 0; }
int32_t bndProcessWMsgs(SBnode *pBnode, SArray *pMsgs) { return 0; } int32_t bndProcessWMsgs(SBnode *pBnode, SArray *pMsgs) { return 0; }
void bndDestroy(const char *path) { taosRemoveDir(path); }

2
source/dnode/mgmt/dnode/inc/dmInt.h
View File

@ -29,7 +29,7 @@ typedef struct SDnodeMgmt {
SEpSet mnodeEpSet; SEpSet mnodeEpSet;
SHashObj *dnodeHash; SHashObj *dnodeHash;
SArray *dnodeEps; SArray *dnodeEps;
pthread_t *threadId; TdThread *threadId;
SRWLatch latch; SRWLatch latch;
SDnodeWorker mgmtWorker; SDnodeWorker mgmtWorker;
SDnodeWorker statusWorker; SDnodeWorker statusWorker;

2
source/dnode/mgmt/dnode/src/dmWorker.c
View File

@ -30,7 +30,7 @@ static void *dmThreadRoutine(void *param) {
setThreadName("dnode-hb"); setThreadName("dnode-hb");
while (true) { while (true) {
pthread_testcancel(); taosThreadTestCancel();
taosMsleep(200); taosMsleep(200);
if (dndGetStatus(pDnode) != DND_STAT_RUNNING || pDnode->dropped) { if (dndGetStatus(pDnode) != DND_STAT_RUNNING || pDnode->dropped) {
continue; continue;

2
source/dnode/mgmt/test/sut/inc/server.h
View File

@ -28,7 +28,7 @@ class TestServer {
private: private:
SDnode* pDnode; SDnode* pDnode;
TdThread* threadId; TdThread threadId;
char path[PATH_MAX]; char path[PATH_MAX];
char fqdn[TSDB_FQDN_LEN]; char fqdn[TSDB_FQDN_LEN];
char firstEp[TSDB_EP_LEN]; char firstEp[TSDB_EP_LEN];

12
source/dnode/mgmt/test/sut/src/server.cpp
View File

@ -41,11 +41,11 @@ bool TestServer::DoStart() {
return false; return false;
} }
pthread_attr_t thAttr; TdThreadAttr thAttr;
pthread_attr_init(&thAttr); taosThreadAttrInit(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
pthread_create(&threadId, &thAttr, serverLoop, pDnode); taosThreadCreate(&threadId, &thAttr, serverLoop, pDnode);
pthread_attr_destroy(&thAttr); taosThreadAttrDestroy(&thAttr);
taosMsleep(2100); taosMsleep(2100);
return true; return true;
} }
@ -69,7 +69,7 @@ bool TestServer::Start(const char* path, const char* fqdn, uint16_t port, const
void TestServer::Stop() { void TestServer::Stop() {
dndHandleEvent(pDnode, DND_EVENT_STOP); dndHandleEvent(pDnode, DND_EVENT_STOP);
pthread_join(threadId, NULL); taosThreadJoin(threadId, NULL);
if (pDnode != NULL) { if (pDnode != NULL) {
dndClose(pDnode); dndClose(pDnode);

2
source/dnode/mgmt/vnode/inc/vmInt.h
View File

@ -71,7 +71,7 @@ typedef struct {
int32_t opened; int32_t opened;
int32_t failed; int32_t failed;
int32_t threadIndex; int32_t threadIndex;
pthread_t thread; TdThread thread;
SVnodesMgmt *pMgmt; SVnodesMgmt *pMgmt;
SWrapperCfg *pCfgs; SWrapperCfg *pCfgs;
} SVnodeThread; } SVnodeThread;

12
source/dnode/mgmt/vnode/src/vmInt.c
View File

@ -193,20 +193,20 @@ static int32_t vmOpenVnodes(SVnodesMgmt *pMgmt) {
SVnodeThread *pThread = &threads[t]; SVnodeThread *pThread = &threads[t];
if (pThread->vnodeNum == 0) continue; if (pThread->vnodeNum == 0) continue;
pthread_attr_t thAttr; TdThreadAttr thAttr;
pthread_attr_init(&thAttr); taosThreadAttrInit(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&pThread->thread, &thAttr, vmOpenVnodeFunc, pThread) != 0) { if (taosThreadCreate(&pThread->thread, &thAttr, vmOpenVnodeFunc, pThread) != 0) {
dError("thread:%d, failed to create thread to open vnode, reason:%s", pThread->threadIndex, strerror(errno)); dError("thread:%d, failed to create thread to open vnode, reason:%s", pThread->threadIndex, strerror(errno));
} }
pthread_attr_destroy(&thAttr); taosThreadAttrDestroy(&thAttr);
} }
for (int32_t t = 0; t < threadNum; ++t) { for (int32_t t = 0; t < threadNum; ++t) {
SVnodeThread *pThread = &threads[t]; SVnodeThread *pThread = &threads[t];
if (pThread->vnodeNum > 0 && taosCheckPthreadValid(pThread->thread)) { if (pThread->vnodeNum > 0 && taosCheckPthreadValid(pThread->thread)) {
pthread_join(pThread->thread, NULL); taosThreadJoin(pThread->thread, NULL);
} }
free(pThread->pCfgs); free(pThread->pCfgs);
} }

50
source/util/src/tprocess.c
View File

@ -41,16 +41,16 @@ typedef struct SProcQueue {
ProcConsumeFp consumeFp; ProcConsumeFp consumeFp;
void *pParent; void *pParent;
tsem_t sem; tsem_t sem;
pthread_mutex_t *mutex; TdThreadMutex *mutex;
int32_t mutexShmid; int32_t mutexShmid;
int32_t bufferShmid; int32_t bufferShmid;
const char *name; const char *name;
} SProcQueue; } SProcQueue;
typedef struct SProcObj { typedef struct SProcObj {
pthread_t childThread; TdThread childThread;
SProcQueue *pChildQueue; SProcQueue *pChildQueue;
pthread_t parentThread; TdThread parentThread;
SProcQueue *pParentQueue; SProcQueue *pParentQueue;
const char *name; const char *name;
int32_t pid; int32_t pid;
@ -59,9 +59,9 @@ typedef struct SProcObj {
bool testFlag; bool testFlag;
} SProcObj; } SProcObj;
static int32_t taosProcInitMutex(pthread_mutex_t **ppMutex, int32_t *pShmid) { static int32_t taosProcInitMutex(TdThreadMutex **ppMutex, int32_t *pShmid) {
pthread_mutex_t *pMutex = NULL; TdThreadMutex *pMutex = NULL;
pthread_mutexattr_t mattr = {0}; TdThreadMutexAttr mattr = {0};
int32_t shmid = -1; int32_t shmid = -1;
int32_t code = -1; int32_t code = -1;
@ -77,21 +77,21 @@ static int32_t taosProcInitMutex(pthread_mutex_t **ppMutex, int32_t *pShmid) {
goto _OVER; goto _OVER;
} }
shmid = shmget(IPC_PRIVATE, sizeof(pthread_mutex_t), 0600); shmid = shmget(IPC_PRIVATE, sizeof(TdThreadMutex), 0600);
if (shmid <= 0) { if (shmid <= 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to init mutex while shmget since %s", terrstr()); uError("failed to init mutex while shmget since %s", terrstr());
goto _OVER; goto _OVER;
} }
pMutex = (pthread_mutex_t *)shmat(shmid, NULL, 0); pMutex = (TdThreadMutex *)shmat(shmid, NULL, 0);
if (pMutex == NULL) { if (pMutex == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to init mutex while shmat since %s", terrstr()); uError("failed to init mutex while shmat since %s", terrstr());
goto _OVER; goto _OVER;
} }
if (pthread_mutex_init(pMutex, &mattr) != 0) { if (taosThreadMutexInit(pMutex, &mattr) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to init mutex since %s", terrstr()); uError("failed to init mutex since %s", terrstr());
goto _OVER; goto _OVER;
@ -101,7 +101,7 @@ static int32_t taosProcInitMutex(pthread_mutex_t **ppMutex, int32_t *pShmid) {
_OVER: _OVER:
if (code != 0) { if (code != 0) {
pthread_mutex_destroy(pMutex); taosThreadMutexDestroy(pMutex);
shmctl(shmid, IPC_RMID, NULL); shmctl(shmid, IPC_RMID, NULL);
} else { } else {
*ppMutex = pMutex; *ppMutex = pMutex;
@ -112,9 +112,9 @@ _OVER:
return code; return code;
} }
static void taosProcDestroyMutex(pthread_mutex_t *pMutex, int32_t *pShmid) { static void taosProcDestroyMutex(TdThreadMutex *pMutex, int32_t *pShmid) {
if (pMutex != NULL) { if (pMutex != NULL) {
pthread_mutex_destroy(pMutex); taosThreadMutexDestroy(pMutex);
} }
if (*pShmid > 0) { if (*pShmid > 0) {
shmctl(*pShmid, IPC_RMID, NULL); shmctl(*pShmid, IPC_RMID, NULL);
@ -129,7 +129,7 @@ static int32_t taosProcInitBuffer(void **ppBuffer, int32_t size) {
return -1; return -1;
} }
void *shmptr = (pthread_mutex_t *)shmat(shmid, NULL, 0); void *shmptr = shmat(shmid, NULL, 0);
if (shmptr == NULL) { if (shmptr == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to init buffer while shmat since %s", terrstr()); uError("failed to init buffer while shmat since %s", terrstr());
@ -204,9 +204,9 @@ static int32_t taosProcQueuePush(SProcQueue *pQueue, char *pHead, int32_t rawHea
const int32_t bodyLen = CEIL8(rawBodyLen); const int32_t bodyLen = CEIL8(rawBodyLen);
const int32_t fullLen = headLen + bodyLen + 8; const int32_t fullLen = headLen + bodyLen + 8;
pthread_mutex_lock(pQueue->mutex); taosThreadMutexLock(pQueue->mutex);
if (fullLen > pQueue->avail) { if (fullLen > pQueue->avail) {
pthread_mutex_unlock(pQueue->mutex); taosThreadMutexUnlock(pQueue->mutex);
terrno = TSDB_CODE_OUT_OF_SHM_MEM; terrno = TSDB_CODE_OUT_OF_SHM_MEM;
return -1; return -1;
} }
@ -252,7 +252,7 @@ static int32_t taosProcQueuePush(SProcQueue *pQueue, char *pHead, int32_t rawHea
pQueue->avail -= fullLen; pQueue->avail -= fullLen;
pQueue->items++; pQueue->items++;
pthread_mutex_unlock(pQueue->mutex); taosThreadMutexUnlock(pQueue->mutex);
tsem_post(&pQueue->sem); tsem_post(&pQueue->sem);
uTrace("proc:%s, push msg:%p:%d cont:%p:%d to queue:%p", pQueue->name, pHead, rawHeadLen, pBody, rawBodyLen, pQueue); uTrace("proc:%s, push msg:%p:%d cont:%p:%d to queue:%p", pQueue->name, pHead, rawHeadLen, pBody, rawBodyLen, pQueue);
@ -263,9 +263,9 @@ static int32_t taosProcQueuePop(SProcQueue *pQueue, void **ppHead, int32_t *pHea
int32_t *pBodyLen) { int32_t *pBodyLen) {
tsem_wait(&pQueue->sem); tsem_wait(&pQueue->sem);
pthread_mutex_lock(pQueue->mutex); taosThreadMutexLock(pQueue->mutex);
if (pQueue->total - pQueue->avail <= 0) { if (pQueue->total - pQueue->avail <= 0) {
pthread_mutex_unlock(pQueue->mutex); taosThreadMutexUnlock(pQueue->mutex);
tsem_post(&pQueue->sem); tsem_post(&pQueue->sem);
terrno = TSDB_CODE_OUT_OF_SHM_MEM; terrno = TSDB_CODE_OUT_OF_SHM_MEM;
return -1; return -1;
@ -284,7 +284,7 @@ static int32_t taosProcQueuePop(SProcQueue *pQueue, void **ppHead, int32_t *pHea
void *pHead = (*pQueue->mallocHeadFp)(headLen); void *pHead = (*pQueue->mallocHeadFp)(headLen);
void *pBody = (*pQueue->mallocBodyFp)(bodyLen); void *pBody = (*pQueue->mallocBodyFp)(bodyLen);
if (pHead == NULL || pBody == NULL) { if (pHead == NULL || pBody == NULL) {
pthread_mutex_unlock(pQueue->mutex); taosThreadMutexUnlock(pQueue->mutex);
tsem_post(&pQueue->sem); tsem_post(&pQueue->sem);
(*pQueue->freeHeadFp)(pHead); (*pQueue->freeHeadFp)(pHead);
(*pQueue->freeBodyFp)(pBody); (*pQueue->freeBodyFp)(pBody);
@ -325,7 +325,7 @@ static int32_t taosProcQueuePop(SProcQueue *pQueue, void **ppHead, int32_t *pHea
pQueue->avail = pQueue->avail + headLen + bodyLen + 8; pQueue->avail = pQueue->avail + headLen + bodyLen + 8;
pQueue->items--; pQueue->items--;
pthread_mutex_unlock(pQueue->mutex); taosThreadMutexUnlock(pQueue->mutex);
*ppHead = pHead; *ppHead = pHead;
*ppBody = pBody; *ppBody = pBody;
@ -409,12 +409,12 @@ static void taosProcThreadLoop(SProcQueue *pQueue) {
} }
int32_t taosProcRun(SProcObj *pProc) { int32_t taosProcRun(SProcObj *pProc) {
pthread_attr_t thAttr = {0}; TdThreadAttr thAttr;
pthread_attr_init(&thAttr); taosThreadAttrInit(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pProc->isChild || pProc->testFlag) { if (pProc->isChild || pProc->testFlag) {
if (pthread_create(&pProc->childThread, &thAttr, (ProcThreadFp)taosProcThreadLoop, pProc->pChildQueue) != 0) { if (taosThreadCreate(&pProc->childThread, &thAttr, (ProcThreadFp)taosProcThreadLoop, pProc->pChildQueue) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to create thread since %s", terrstr()); uError("failed to create thread since %s", terrstr());
return -1; return -1;
@ -423,7 +423,7 @@ int32_t taosProcRun(SProcObj *pProc) {
} }
if (!pProc->isChild || pProc->testFlag) { if (!pProc->isChild || pProc->testFlag) {
if (pthread_create(&pProc->parentThread, &thAttr, (ProcThreadFp)taosProcThreadLoop, pProc->pParentQueue) != 0) { if (taosThreadCreate(&pProc->parentThread, &thAttr, (ProcThreadFp)taosProcThreadLoop, pProc->pParentQueue) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to create thread since %s", terrstr()); uError("failed to create thread since %s", terrstr());
return -1; return -1;

13
source/util/test/queueTest.cpp
View File

@ -27,6 +27,7 @@ class UtilTestQueue : public ::testing::Test {
static void TearDownTestSuite() {} static void TearDownTestSuite() {}
}; };
#if 0
TEST_F(UtilTestQueue, 01_fork) { TEST_F(UtilTestQueue, 01_fork) {
pid_t pid; pid_t pid;
int shmid; int shmid;
@ -84,7 +85,7 @@ TEST_F(UtilTestQueue, 01_fork) {
perror("fork error"); perror("fork error");
exit(1); exit(1);
} else if (pid == 0) { } else if (pid == 0) {
if ((err = pthread_mutex_lock(m)) < 0) { if ((err = taosThreadMutexLock(m)) < 0) {
printf("lock error:%s\n", strerror(err)); printf("lock error:%s\n", strerror(err));
exit(1); exit(1);
} }
@ -93,14 +94,14 @@ TEST_F(UtilTestQueue, 01_fork) {
(*shmptr2)++; (*shmptr2)++;
} }
if ((err = pthread_mutex_unlock(m)) < 0) { if ((err = taosThreadMutexUnlock(m)) < 0) {
printf("unlock error:%s\n", strerror(err)); printf("unlock error:%s\n", strerror(err));
exit(1); exit(1);
} }
exit(0); exit(0);
} else { } else {
if ((err = pthread_mutex_lock(m)) < 0) { if ((err = taosThreadMutexLock(m)) < 0) {
printf("lock error:%s\n", strerror(err)); printf("lock error:%s\n", strerror(err));
exit(1); exit(1);
} }
@ -108,7 +109,7 @@ TEST_F(UtilTestQueue, 01_fork) {
**shmptr2 = i; **shmptr2 = i;
(*shmptr2)++; (*shmptr2)++;
} }
if ((err = pthread_mutex_unlock(m)) < 0) { if ((err = taosThreadMutexUnlock(m)) < 0) {
printf("unlock error:%s\n", strerror(err)); printf("unlock error:%s\n", strerror(err));
exit(1); exit(1);
} }
@ -122,9 +123,11 @@ TEST_F(UtilTestQueue, 01_fork) {
printf("\n"); printf("\n");
pthread_mutexattr_destroy(&mattr); taosThreadAttrDestroy(&mattr);
//销毁mutex //销毁mutex
pthread_mutex_destroy(m); pthread_mutex_destroy(m);
exit(0); exit(0);
} }
#endif