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

fix: restore tdb page cache lock

This commit is contained in:
dapan1121 2023-04-20 18:56:18 +08:00
parent 36f3e44096
commit fa77356cc4
1 changed files with 10 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
source/libs/tdb/src/db/tdbPCache.c
View File

@ -21,7 +21,7 @@ struct SPCache {
int szPage; int szPage;
int nPages; int nPages;
SPage **aPage; SPage **aPage;
TdThreadRwlock rwLock; tdb_mutex_t mutex;
int nFree; int nFree;
SPage *pFree; SPage *pFree;
int nPage; int nPage;
@ -44,11 +44,10 @@ static void tdbPCacheAddPageToHash(SPCache *pCache, SPage *pPage);
static void tdbPCacheUnpinPage(SPCache *pCache, SPage *pPage); static void tdbPCacheUnpinPage(SPCache *pCache, SPage *pPage);
static int tdbPCacheCloseImpl(SPCache *pCache); static int tdbPCacheCloseImpl(SPCache *pCache);
static void tdbPCacheInitLock(SPCache *pCache) { taosThreadRwlockInit(&(pCache->rwLock), NULL); } static void tdbPCacheInitLock(SPCache *pCache) { tdbMutexInit(&(pCache->mutex), NULL); }
static void tdbPCacheDestroyLock(SPCache *pCache) { taosThreadRwlockDestroy(&(pCache->rwLock)); } static void tdbPCacheDestroyLock(SPCache *pCache) { tdbMutexDestroy(&(pCache->mutex)); }
static void tdbPCacheRLock(SPCache *pCache) { taosThreadRwlockRdlock(&(pCache->rwLock)); } static void tdbPCacheLock(SPCache *pCache) { tdbMutexLock(&(pCache->mutex)); }
static void tdbPCacheWLock(SPCache *pCache) { taosThreadRwlockWrlock(&(pCache->rwLock)); } static void tdbPCacheUnlock(SPCache *pCache) { tdbMutexUnlock(&(pCache->mutex)); }
static void tdbPCacheUnlock(SPCache *pCache) { taosThreadRwlockUnlock(&(pCache->rwLock)); }
int tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache) { int tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache) {
SPCache *pCache; SPCache *pCache;
@ -159,7 +158,7 @@ static int tdbPCacheAlterImpl(SPCache *pCache, int32_t nPage) {
int tdbPCacheAlter(SPCache *pCache, int32_t nPage) { int tdbPCacheAlter(SPCache *pCache, int32_t nPage) {
int ret = 0; int ret = 0;
tdbPCacheWLock(pCache); tdbPCacheLock(pCache);
ret = tdbPCacheAlterImpl(pCache, nPage); ret = tdbPCacheAlterImpl(pCache, nPage);
@ -172,6 +171,8 @@ SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, TXN *pTxn) {
SPage *pPage; SPage *pPage;
i32 nRef = 0; i32 nRef = 0;
tdbPCacheLock(pCache);
pPage = tdbPCacheFetchImpl(pCache, pPgid, pTxn); pPage = tdbPCacheFetchImpl(pCache, pPgid, pTxn);
if (pPage) { if (pPage) {
nRef = tdbRefPage(pPage); nRef = tdbRefPage(pPage);
@ -192,7 +193,7 @@ SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, TXN *pTxn) {
} }
void tdbPCacheMarkFree(SPCache *pCache, SPage *pPage) { void tdbPCacheMarkFree(SPCache *pCache, SPage *pPage) {
tdbPCacheWLock(pCache); tdbPCacheLock(pCache);
tdbPCacheRemovePageFromHash(pCache, pPage); tdbPCacheRemovePageFromHash(pCache, pPage);
pPage->isFree = 1; pPage->isFree = 1;
tdbPCacheUnlock(pCache); tdbPCacheUnlock(pCache);
@ -240,7 +241,7 @@ void tdbPCacheRelease(SPCache *pCache, SPage *pPage, TXN *pTxn) {
return; return;
} }
tdbPCacheWLock(pCache); tdbPCacheLock(pCache);
nRef = tdbUnrefPage(pPage); nRef = tdbUnrefPage(pPage);
tdbTrace("pcache/release page %p/%d/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id, nRef); tdbTrace("pcache/release page %p/%d/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id, nRef);
if (nRef == 0) { if (nRef == 0) {
@ -274,8 +275,6 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, TXN *pTxn)
SPage *pPage = NULL; SPage *pPage = NULL;
SPage *pPageH = NULL; SPage *pPageH = NULL;
tdbPCacheRLock(pCache);
if (!pTxn) { if (!pTxn) {
tdbError("tdb/pcache: null ptr pTxn, fetch impl failed."); tdbError("tdb/pcache: null ptr pTxn, fetch impl failed.");
return NULL; return NULL;
@ -295,10 +294,6 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, TXN *pTxn)
} }
} }
tdbPCacheUnlock(pCache);
tdbPCacheWLock(pCache);
// 1. pPage == NULL // 1. pPage == NULL
// 2. pPage && pPage->isLocal == 0 && !TDB_TXN_IS_WRITE(pTxn) // 2. pPage && pPage->isLocal == 0 && !TDB_TXN_IS_WRITE(pTxn)
pPageH = pPage; pPageH = pPage;