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

[TD-3353]<hotfix>: solve race condition coredump

This commit is contained in:
Hongze Cheng 2021-03-18 15:50:55 +08:00
parent 0c46529296
commit e4e46768a6
5 changed files with 82 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/inc/tsdb.h
View File

@ -25,6 +25,8 @@
#include "tdataformat.h"
#include "tname.h"
#include "hash.h"
#include "tlockfree.h"
#include "tlist.h"
#ifdef __cplusplus
extern "C" {
@ -172,10 +174,32 @@ typedef struct STsdbQueryCond {
int32_t type; // data block load type:
} STsdbQueryCond;
typedef struct STableData STableData;
typedef struct {
T_REF_DECLARE()
SRWLatch latch;
TSKEY keyFirst;
TSKEY keyLast;
int64_t numOfRows;
int32_t maxTables;
STableData **tData;
SList * actList;
SList * extraBuffList;
SList * bufBlockList;
int64_t pointsAdd; // TODO
int64_t storageAdd; // TODO
} SMemTable;
typedef struct {
SMemTable* mem;
SMemTable* imem;
SMemTable mtable;
SMemTable* omem;
} SMemSnapshot;
typedef struct SMemRef {
int32_t ref;
void * mem;
void * imem;
int32_t ref;
SMemSnapshot snapshot;
} SMemRef;
typedef struct SDataBlockInfo {

2
src/query/src/qExecutor.c
View File

@ -1840,7 +1840,7 @@ static void doFreeQueryHandle(SQueryRuntimeEnv* pRuntimeEnv) {
pRuntimeEnv->pQueryHandle = NULL;
SMemRef* pMemRef = &pQuery->memRef;
assert(pMemRef->ref == 0 && pMemRef->imem == NULL && pMemRef->mem == NULL);
assert(pMemRef->ref == 0 && pMemRef->snapshot.imem == NULL && pMemRef->snapshot.mem == NULL);
}
static void teardownQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv) {

23
src/tsdb/inc/tsdbMemTable.h
View File

@ -31,29 +31,14 @@ typedef struct {
SSkipListIterator *pIter;
} SCommitIter;
typedef struct {
struct STableData {
uint64_t uid;
TSKEY keyFirst;
TSKEY keyLast;
int64_t numOfRows;
SSkipList* pData;
T_REF_DECLARE()
} STableData;
typedef struct {
T_REF_DECLARE()
SRWLatch latch;
TSKEY keyFirst;
TSKEY keyLast;
int64_t numOfRows;
int32_t maxTables;
STableData** tData;
SList* actList;
SList* extraBuffList;
SList* bufBlockList;
int64_t pointsAdd; // TODO
int64_t storageAdd; // TODO
} SMemTable;
};
enum { TSDB_UPDATE_META, TSDB_DROP_META };
@ -77,8 +62,8 @@ typedef struct {
int tsdbRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
int tsdbUnRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
int tsdbTakeMemSnapshot(STsdbRepo* pRepo, SMemTable** pMem, SMemTable** pIMem, SArray* pATable);
void tsdbUnTakeMemSnapShot(STsdbRepo* pRepo, SMemTable* pMem, SMemTable* pIMem);
int tsdbTakeMemSnapshot(STsdbRepo* pRepo, SMemSnapshot* pSnapshot, SArray* pATable);
void tsdbUnTakeMemSnapShot(STsdbRepo* pRepo, SMemSnapshot* pSnapshot);
void* tsdbAllocBytes(STsdbRepo* pRepo, int bytes);
int tsdbAsyncCommit(STsdbRepo* pRepo);
int tsdbLoadDataFromCache(STable* pTable, SSkipListIterator* pIter, TSKEY maxKey, int maxRowsToRead, SDataCols* pCols,

92
src/tsdb/src/tsdbMemTable.c
View File

@ -124,88 +124,80 @@ int tsdbUnRefMemTable(STsdbRepo *pRepo, SMemTable *pMemTable) {
return 0;
}
int tsdbTakeMemSnapshot(STsdbRepo *pRepo, SMemTable **pMem, SMemTable **pIMem, SArray *pATable) {
SMemTable *tmem;
int tsdbTakeMemSnapshot(STsdbRepo *pRepo, SMemSnapshot *pSnapshot, SArray *pATable) {
memset(pSnapshot, 0, sizeof(*pSnapshot));
// Get snap object
if (tsdbLockRepo(pRepo) < 0) return -1;
tmem = pRepo->mem;
*pIMem = pRepo->imem;
tsdbRefMemTable(pRepo, tmem);
tsdbRefMemTable(pRepo, *pIMem);
pSnapshot->omem = pRepo->mem;
pSnapshot->imem = pRepo->imem;
tsdbRefMemTable(pRepo, pRepo->mem);
tsdbRefMemTable(pRepo, pRepo->imem);
if (tsdbUnlockRepo(pRepo) < 0) return -1;
// Copy mem objects and ref needed STableData
if (tmem) {
taosRLockLatch(&(tmem->latch));
if (pSnapshot->omem) {
taosRLockLatch(&(pSnapshot->omem->latch));
*pMem = (SMemTable *)calloc(1, sizeof(**pMem));
if (*pMem == NULL) {
pSnapshot->mem = &(pSnapshot->mtable);
pSnapshot->mem->tData = (STableData **)calloc(pSnapshot->omem->maxTables, sizeof(STableData *));
if (pSnapshot->mem->tData == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
taosRUnLockLatch(&(tmem->latch));
tsdbUnRefMemTable(pRepo, tmem);
tsdbUnRefMemTable(pRepo, *pIMem);
*pMem = NULL;
*pIMem = NULL;
taosRUnLockLatch(&(pSnapshot->omem->latch));
tsdbUnRefMemTable(pRepo, pSnapshot->omem);
tsdbUnRefMemTable(pRepo, pSnapshot->imem);
pSnapshot->mem = NULL;
pSnapshot->imem = NULL;
pSnapshot->omem = NULL;
return -1;
}
(*pMem)->tData = (STableData **)calloc(tmem->maxTables, sizeof(STableData *));
if ((*pMem)->tData == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
taosRUnLockLatch(&(tmem->latch));
free(*pMem);
tsdbUnRefMemTable(pRepo, tmem);
tsdbUnRefMemTable(pRepo, *pIMem);
*pMem = NULL;
*pIMem = NULL;
return -1;
}
(*pMem)->keyFirst = tmem->keyFirst;
(*pMem)->keyLast = tmem->keyLast;
(*pMem)->numOfRows = tmem->numOfRows;
(*pMem)->maxTables = tmem->maxTables;
pSnapshot->mem->keyFirst = pSnapshot->omem->keyFirst;
pSnapshot->mem->keyLast = pSnapshot->omem->keyLast;
pSnapshot->mem->numOfRows = pSnapshot->omem->numOfRows;
pSnapshot->mem->maxTables = pSnapshot->omem->maxTables;
for (size_t i = 0; i < taosArrayGetSize(pATable); i++) {
STable * pTable = *(STable **)taosArrayGet(pATable, i);
int32_t tid = TABLE_TID(pTable);
STableData *pTableData = (tid < tmem->maxTables) ? tmem->tData[tid] : NULL;
STableData *pTableData = (tid < pSnapshot->omem->maxTables) ? pSnapshot->omem->tData[tid] : NULL;
if ((pTableData == NULL) || (TABLE_UID(pTable) != pTableData->uid)) continue;
(*pMem)->tData[tid] = tmem->tData[tid];
T_REF_INC(tmem->tData[tid]);
pSnapshot->mem->tData[tid] = pTableData;
T_REF_INC(pTableData);
}
taosRUnLockLatch(&(tmem->latch));
taosRUnLockLatch(&(pSnapshot->omem->latch));
}
tsdbUnRefMemTable(pRepo, tmem);
tsdbDebug("vgId:%d take memory snapshot, pMem %p pIMem %p", REPO_ID(pRepo), *pMem, *pIMem);
tsdbDebug("vgId:%d take memory snapshot, pMem %p pIMem %p", REPO_ID(pRepo), pSnapshot->omem, pSnapshot->imem);
return 0;
}
void tsdbUnTakeMemSnapShot(STsdbRepo *pRepo, SMemTable *pMem, SMemTable *pIMem) {
tsdbDebug("vgId:%d untake memory snapshot, pMem %p pIMem %p", REPO_ID(pRepo), pMem, pIMem);
void tsdbUnTakeMemSnapShot(STsdbRepo *pRepo, SMemSnapshot *pSnapshot) {
tsdbDebug("vgId:%d untake memory snapshot, pMem %p pIMem %p", REPO_ID(pRepo), pSnapshot->omem, pSnapshot->imem);
if (pMem != NULL) {
for (size_t i = 0; i < pMem->maxTables; i++) {
STableData *pTableData = pMem->tData[i];
if (pSnapshot->mem) {
ASSERT(pSnapshot->omem != NULL);
for (size_t i = 0; i < pSnapshot->mem->maxTables; i++) {
STableData *pTableData = pSnapshot->mem->tData[i];
if (pTableData) {
tsdbFreeTableData(pTableData);
}
}
free(pMem->tData);
free(pMem);
tfree(pSnapshot->mem->tData);
tsdbUnRefMemTable(pRepo, pSnapshot->omem);
}
if (pIMem != NULL) {
tsdbUnRefMemTable(pRepo, pIMem);
}
tsdbUnRefMemTable(pRepo, pSnapshot->imem);
pSnapshot->mem = NULL;
pSnapshot->imem = NULL;
pSnapshot->omem = NULL;
}
void *tsdbAllocBytes(STsdbRepo *pRepo, int bytes) {

18
src/tsdb/src/tsdbRead.c
View File

@ -194,7 +194,7 @@ static void tsdbMayTakeMemSnapshot(STsdbQueryHandle* pQueryHandle, SArray* psTab
SMemRef* pMemRef = pQueryHandle->pMemRef;
if (pQueryHandle->pMemRef->ref++ == 0) {
tsdbTakeMemSnapshot(pQueryHandle->pTsdb, (SMemTable**)&(pMemRef->mem), (SMemTable**)&(pMemRef->imem), psTable);
tsdbTakeMemSnapshot(pQueryHandle->pTsdb, &(pMemRef->snapshot), psTable);
}
taosArrayDestroy(psTable);
@ -208,9 +208,7 @@ static void tsdbMayUnTakeMemSnapshot(STsdbQueryHandle* pQueryHandle) {
}
if (--pMemRef->ref == 0) {
tsdbUnTakeMemSnapShot(pQueryHandle->pTsdb, pMemRef->mem, pMemRef->imem);
pMemRef->mem = NULL;
pMemRef->imem = NULL;
tsdbUnTakeMemSnapShot(pQueryHandle->pTsdb, &(pMemRef->snapshot));
}
pQueryHandle->pMemRef = NULL;
@ -229,10 +227,10 @@ int64_t tsdbGetNumOfRowsInMemTable(TsdbQueryHandleT* pHandle) {
if (pMemRef == NULL) { return rows; }
STableData* pMem = NULL;
STableData* pIMem = NULL;
STableData* pIMem = NULL;
SMemTable *pMemT = (SMemTable *)(pMemRef->mem);
SMemTable *pIMemT = (SMemTable *)(pMemRef->imem);
SMemTable* pMemT = pMemRef->snapshot.mem;
SMemTable* pIMemT = pMemRef->snapshot.imem;
if (pMemT && pCheckInfo->tableId.tid < pMemT->maxTables) {
pMem = pMemT->tData[pCheckInfo->tableId.tid];
@ -605,7 +603,7 @@ static bool initTableMemIterator(STsdbQueryHandle* pHandle, STableCheckInfo* pCh
int32_t order = pHandle->order;
// no data in buffer, abort
if (pHandle->pMemRef->mem == NULL && pHandle->pMemRef->imem == NULL) {
if (pHandle->pMemRef->snapshot.mem == NULL && pHandle->pMemRef->snapshot.imem == NULL) {
return false;
}
@ -614,8 +612,8 @@ static bool initTableMemIterator(STsdbQueryHandle* pHandle, STableCheckInfo* pCh
STableData* pMem = NULL;
STableData* pIMem = NULL;
SMemTable* pMemT = pHandle->pMemRef->mem;
SMemTable* pIMemT = pHandle->pMemRef->imem;
SMemTable* pMemT = pHandle->pMemRef->snapshot.mem;
SMemTable* pIMemT = pHandle->pMemRef->snapshot.imem;
if (pMemT && pCheckInfo->tableId.tid < pMemT->maxTables) {
pMem = pMemT->tData[pCheckInfo->tableId.tid];