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

fix: r/w concurrency

This commit is contained in:
Hongze Cheng 2022-07-19 08:30:49 +00:00
parent f3204ed912
commit 706b081f70
6 changed files with 68 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
source/dnode/vnode/src/inc/vnd.h
View File

@ -62,12 +62,13 @@ struct SVBufPoolNode {
};
struct SVBufPool {
SVBufPool* next;
int64_t nRef;
int64_t size;
uint8_t* ptr;
SVBufPoolNode* pTail;
SVBufPoolNode node;
SVBufPool* next;
SVnode* pVnode;
volatile int32_t nRef;
int64_t size;
uint8_t* ptr;
SVBufPoolNode* pTail;
SVBufPoolNode node;
};
int32_t vnodeOpenBufPool(SVnode* pVnode, int64_t size);
@ -78,7 +79,7 @@ void vnodeBufPoolReset(SVBufPool* pPool);
int32_t vnodeQueryOpen(SVnode* pVnode);
void vnodeQueryClose(SVnode* pVnode);
int32_t vnodeGetTableMeta(SVnode* pVnode, SRpcMsg* pMsg);
int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg);
int vnodeGetTableCfg(SVnode* pVnode, SRpcMsg* pMsg);
// vnodeCommit.c
int32_t vnodeBegin(SVnode* pVnode);

42
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -77,6 +77,8 @@ typedef struct SSnapDataHdr SSnapDataHdr;
// vnd.h
void* vnodeBufPoolMalloc(SVBufPool* pPool, int size);
void vnodeBufPoolFree(SVBufPool* pPool, void* p);
void vnodeBufPoolRef(SVBufPool* pPool);
void vnodeBufPoolUnRef(SVBufPool* pPool);
// meta
typedef struct SMCtbCursor SMCtbCursor;
@ -247,26 +249,26 @@ struct STsdbKeepCfg {
};
struct SVnode {
char* path;
SVnodeCfg config;
SVState state;
STfs* pTfs;
SMsgCb msgCb;
SVBufPool* pPool;
SVBufPool* inUse;
SVBufPool* onCommit;
SVBufPool* onRecycle;
SMeta* pMeta;
SSma* pSma;
STsdb* pTsdb;
SWal* pWal;
STQ* pTq;
SSink* pSink;
tsem_t canCommit;
int64_t sync;
int32_t blockCount;
tsem_t syncSem;
SQHandle* pQuery;
char* path;
SVnodeCfg config;
SVState state;
STfs* pTfs;
SMsgCb msgCb;
TdThreadMutex mutex;
TdThreadCond poolNotEmpty;
SVBufPool* pPool;
SVBufPool* inUse;
SMeta* pMeta;
SSma* pSma;
STsdb* pTsdb;
SWal* pWal;
STQ* pTq;
SSink* pSink;
tsem_t canCommit;
int64_t sync;
int32_t blockCount;
tsem_t syncSem;
SQHandle* pQuery;
};
#define TD_VID(PVNODE) ((PVNODE)->config.vgId)

2
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -55,6 +55,7 @@ int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable) {
taosMemoryFree(pMemTable);
goto _err;
}
vnodeBufPoolRef(pMemTable->pPool);
*ppMemTable = pMemTable;
return code;
@ -66,6 +67,7 @@ _err:
void tsdbMemTableDestroy(SMemTable *pMemTable) {
if (pMemTable) {
vnodeBufPoolUnRef(pMemTable->pPool);
taosArrayDestroy(pMemTable->aTbData);
taosMemoryFree(pMemTable);
}

29
source/dnode/vnode/src/vnd/vnodeBufPool.c
View File

@ -17,7 +17,7 @@
/* ------------------------ STRUCTURES ------------------------ */
static int vnodeBufPoolCreate(int64_t size, SVBufPool **ppPool);
static int vnodeBufPoolCreate(SVnode *pVnode, int64_t size, SVBufPool **ppPool);
static int vnodeBufPoolDestroy(SVBufPool *pPool);
int vnodeOpenBufPool(SVnode *pVnode, int64_t size) {
@ -28,7 +28,7 @@ int vnodeOpenBufPool(SVnode *pVnode, int64_t size) {
for (int i = 0; i < 3; i++) {
// create pool
ret = vnodeBufPoolCreate(size, &pPool);
ret = vnodeBufPoolCreate(pVnode, size, &pPool);
if (ret < 0) {
vError("vgId:%d, failed to open vnode buffer pool since %s", TD_VID(pVnode), tstrerror(terrno));
vnodeCloseBufPool(pVnode);
@ -120,7 +120,7 @@ void vnodeBufPoolFree(SVBufPool *pPool, void *p) {
}
// STATIC METHODS -------------------
static int vnodeBufPoolCreate(int64_t size, SVBufPool **ppPool) {
static int vnodeBufPoolCreate(SVnode *pVnode, int64_t size, SVBufPool **ppPool) {
SVBufPool *pPool;
pPool = taosMemoryMalloc(sizeof(SVBufPool) + size);
@ -130,6 +130,7 @@ static int vnodeBufPoolCreate(int64_t size, SVBufPool **ppPool) {
}
pPool->next = NULL;
pPool->pVnode = pVnode;
pPool->nRef = 0;
pPool->size = 0;
pPool->ptr = pPool->node.data;
@ -146,4 +147,26 @@ static int vnodeBufPoolDestroy(SVBufPool *pPool) {
vnodeBufPoolReset(pPool);
taosMemoryFree(pPool);
return 0;
}
void vnodeBufPoolRef(SVBufPool *pPool) {
int32_t nRef = atomic_fetch_add_32(&pPool->nRef, 1);
ASSERT(nRef > 0);
}
void vnodeBufPoolUnRef(SVBufPool *pPool) {
int32_t nRef = atomic_sub_fetch_32(&pPool->nRef, 1);
if (nRef == 0) {
SVnode *pVnode = pPool->pVnode;
vnodeBufPoolReset(pPool);
taosThreadMutexLock(&pVnode->mutex);
pPool->next = pVnode->pPool;
pVnode->pPool = pPool;
taosThreadCondSignal(&pVnode->poolNotEmpty);
taosThreadMutexUnlock(&pVnode->mutex);
}
}

16
source/dnode/vnode/src/vnd/vnodeCommit.c
View File

@ -15,7 +15,7 @@
#include "vnd.h"
#define VND_INFO_FNAME "vnode.json"
#define VND_INFO_FNAME "vnode.json"
#define VND_INFO_FNAME_TMP "vnode_tmp.json"
static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData);
@ -27,18 +27,18 @@ static void vnodeWaitCommit(SVnode *pVnode);
int vnodeBegin(SVnode *pVnode) {
// alloc buffer pool
/* pthread_mutex_lock(); */
taosThreadMutexLock(&pVnode->mutex);
while (pVnode->pPool == NULL) {
/* pthread_cond_wait(); */
taosThreadCondWait(&pVnode->poolNotEmpty, &pVnode->mutex);
}
pVnode->inUse = pVnode->pPool;
pVnode->inUse->nRef = 1;
pVnode->pPool = pVnode->inUse->next;
pVnode->inUse->next = NULL;
/* ref pVnode->inUse buffer pool */
/* pthread_mutex_unlock(); */
taosThreadMutexUnlock(&pVnode->mutex);
pVnode->state.commitID++;
// begin meta
@ -217,7 +217,7 @@ int vnodeCommit(SVnode *pVnode) {
vInfo("vgId:%d, start to commit, commit ID:%" PRId64 " version:%" PRId64, TD_VID(pVnode), pVnode->state.commitID,
pVnode->state.applied);
pVnode->onCommit = pVnode->inUse;
vnodeBufPoolUnRef(pVnode->inUse);
pVnode->inUse = NULL;
// save info
@ -284,10 +284,6 @@ int vnodeCommit(SVnode *pVnode) {
// apply the commit (TODO)
walEndSnapshot(pVnode->pWal);
vnodeBufPoolReset(pVnode->onCommit);
pVnode->onCommit->next = pVnode->pPool;
pVnode->pPool = pVnode->onCommit;
pVnode->onCommit = NULL;
vInfo("vgId:%d, commit over", TD_VID(pVnode));

4
source/dnode/vnode/src/vnd/vnodeOpen.c
View File

@ -89,6 +89,8 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
tsem_init(&pVnode->syncSem, 0, 0);
tsem_init(&(pVnode->canCommit), 0, 1);
taosThreadMutexInit(&pVnode->mutex, NULL);
taosThreadCondInit(&pVnode->poolNotEmpty, NULL);
// open buffer pool
if (vnodeOpenBufPool(pVnode, pVnode->config.isHeap ? 0 : pVnode->config.szBuf / 3) < 0) {
@ -195,6 +197,8 @@ void vnodeClose(SVnode *pVnode) {
// destroy handle
tsem_destroy(&(pVnode->canCommit));
tsem_destroy(&pVnode->syncSem);
taosThreadCondDestroy(&pVnode->poolNotEmpty);
taosThreadMutexDestroy(&pVnode->mutex);
taosMemoryFree(pVnode);
}
}