diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h index 5b8d70fb7c127d476566f54623c43783cd5a240a..14173690967ffd26be92cf13a05af6f7508533fe 100644 --- a/include/libs/wal/wal.h +++ b/include/libs/wal/wal.h @@ -77,11 +77,11 @@ typedef struct { } SWalSyncInfo; typedef struct { - int8_t protoVer; int64_t version; - int16_t msgType; + int64_t ingestTs; int32_t bodyLen; - int64_t ingestTs; // not implemented + int16_t msgType; + int8_t protoVer; // sync meta SWalSyncInfo syncMeta; diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c index e259dde29c86a7559d8d4dd5f256a81dc137727b..46de219035a36ae6c40c4e8e1ac8bb4cc777d6f7 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCache.c +++ b/source/dnode/vnode/src/tsdb/tsdbCache.c @@ -46,10 +46,10 @@ void tsdbCloseCache(SLRUCache *pCache) { } } -static void getTableCacheKeyS(tb_uid_t uid, const char *cacheType, char *key, int *len) { - snprintf(key, 30, "%" PRIi64 "%s", uid, cacheType); - *len = strlen(key); -} +/* static void getTableCacheKeyS(tb_uid_t uid, const char *cacheType, char *key, int *len) { */ +/* snprintf(key, 30, "%" PRIi64 "%s", uid, cacheType); */ +/* *len = strlen(key); */ +/* } */ static void getTableCacheKey(tb_uid_t uid, int cacheType, char *key, int *len) { if (cacheType == 0) { // last_row @@ -245,8 +245,6 @@ int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb char key[32] = {0}; int keyLen = 0; - // ((void)(row)); - // getTableCacheKey(uid, "l", key, &keyLen); getTableCacheKey(uid, 1, key, &keyLen); LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen); @@ -323,26 +321,10 @@ static tb_uid_t getTableSuidByUid(tb_uid_t uid, STsdb *pTsdb) { static int32_t getTableDelDataFromDelIdx(SDelFReader *pDelReader, SDelIdx *pDelIdx, SArray *aDelData) { int32_t code = 0; - // SMapData delDataMap; - // SDelData delData; - if (pDelIdx) { - // tMapDataReset(&delDataMap); - - // code = tsdbReadDelData(pDelReader, pDelIdx, &delDataMap, NULL); code = tsdbReadDelData(pDelReader, pDelIdx, aDelData, NULL); - if (code) goto _err; - /* - for (int32_t iDelData = 0; iDelData < delDataMap.nItem; ++iDelData) { - code = tMapDataGetItemByIdx(&delDataMap, iDelData, &delData, tGetDelData); - if (code) goto _err; - - taosArrayPush(aDelData, &delData); - } - */ } -_err: return code; } @@ -444,18 +426,16 @@ typedef struct SFSNextRowIter { SArray *aDFileSet; SDataFReader *pDataFReader; SArray *aBlockIdx; - // SMapData blockIdxMap; - // SBlockIdx blockIdx; - SBlockIdx *pBlockIdx; - SMapData blockMap; - int32_t nBlock; - int32_t iBlock; - SBlock block; - SBlockData blockData; - SBlockData *pBlockData; - int32_t nRow; - int32_t iRow; - TSDBROW row; + SBlockIdx *pBlockIdx; + SMapData blockMap; + int32_t nBlock; + int32_t iBlock; + SBlock block; + SBlockData blockData; + SBlockData *pBlockData; + int32_t nRow; + int32_t iRow; + TSDBROW row; } SFSNextRowIter; static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow) { @@ -629,41 +609,8 @@ typedef struct SMemNextRowIter { } SMemNextRowIter; static int32_t getNextRowFromMem(void *iter, TSDBROW **ppRow) { - // static int32_t getNextRowFromMem(void *iter, SArray *pRowArray) { SMemNextRowIter *state = (SMemNextRowIter *)iter; int32_t code = 0; - /* - if (!state->iterOpened) { - if (state->pMem != NULL) { - tsdbTbDataIterOpen(state->pMem, NULL, 1, &state->iter); - - state->iterOpened = true; - - TSDBROW *pMemRow = tsdbTbDataIterGet(&state->iter); - if (pMemRow) { - state->curRow = pMemRow; - } else { - return code; - } - } else { - return code; - } - } - - taosArrayPush(pRowArray, state->curRow); - while (tsdbTbDataIterNext(&state->iter)) { - TSDBROW *row = tsdbTbDataIterGet(&state->iter); - - if (TSDBROW_TS(row) < TSDBROW_TS(state->curRow)) { - state->curRow = row; - break; - } else { - taosArrayPush(pRowArray, row); - } - } - - return code; - */ switch (state->state) { case SMEMNEXTROW_ENTER: { if (state->pMem != NULL) { @@ -768,10 +715,8 @@ static bool tsdbKeyDeleted(TSDBKEY *key, SArray *pSkyline, int64_t *iSkyline) { } typedef int32_t (*_next_row_fn_t)(void *iter, TSDBROW **ppRow); -// typedef int32_t (*_next_row_fn_t)(void *iter, SArray *pRowArray); typedef int32_t (*_next_row_clear_fn_t)(void *iter); -// typedef struct TsdbNextRowState { typedef struct { TSDBROW *pRow; bool stop; @@ -782,7 +727,6 @@ typedef struct { } TsdbNextRowState; typedef struct { - // STsdb *pTsdb; SArray *pSkyline; int64_t iSkyline; @@ -793,10 +737,8 @@ typedef struct { TSDBROW memRow, imemRow, fsRow; TsdbNextRowState input[3]; - // SMemTable *pMemTable; - // SMemTable *pIMemTable; - STsdbReadSnap *pReadSnap; - STsdb *pTsdb; + STsdbReadSnap *pReadSnap; + STsdb *pTsdb; } CacheNextRowIter; static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTsdb) { @@ -967,7 +909,7 @@ _err: return code; } -static int32_t mergeLastRow2(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRow) { +static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRow) { int32_t code = 0; STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, uid, -1); @@ -978,8 +920,6 @@ static int32_t mergeLastRow2(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppR SArray *pColArray = taosArrayInit(nCol, sizeof(SColVal)); SColVal *pColVal = &(SColVal){0}; - // tb_uid_t suid = getTableSuidByUid(uid, pTsdb); - TSKEY lastRowTs = TSKEY_MAX; CacheNextRowIter iter = {0}; @@ -1066,7 +1006,7 @@ _err: return code; } -static int32_t mergeLast2(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) { +static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) { int32_t code = 0; STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, uid, -1); @@ -1077,8 +1017,6 @@ static int32_t mergeLast2(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) { SArray *pColArray = taosArrayInit(nCol, sizeof(SLastCol)); SColVal *pColVal = &(SColVal){0}; - // tb_uid_t suid = getTableSuidByUid(uid, pTsdb); - TSKEY lastRowTs = TSKEY_MAX; CacheNextRowIter iter = {0}; @@ -1124,12 +1062,7 @@ static int32_t mergeLast2(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) { continue; } } - /* - if ((TSDBROW_TS(pRow) < lastRowTs)) { - // goto build the result ts row - break; - } - */ + // merge into pColArray setNoneCol = false; for (iCol = noneCol; iCol < nCol; ++iCol) { @@ -1139,7 +1072,6 @@ static int32_t mergeLast2(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) { tsdbRowGetColVal(pRow, pTSchema, iCol, pColVal); if ((tColVal->isNone || tColVal->isNull) && (!pColVal->isNone && !pColVal->isNull)) { taosArraySet(pColArray, iCol, &(SLastCol){.ts = rowTs, .colVal = *pColVal}); - //} else if (tColVal->isNone && pColVal->isNone && !setNoneCol) { } else if ((tColVal->isNone || tColVal->isNull) && (pColVal->isNone || pColVal->isNull) && !setNoneCol) { noneCol = iCol; setNoneCol = true; @@ -1148,521 +1080,36 @@ static int32_t mergeLast2(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) { } while (setNoneCol); // build the result ts row here - //*dup = false; if (taosArrayGetSize(pColArray) <= 0) { *ppLastArray = NULL; taosArrayDestroy(pColArray); } else { *ppLastArray = pColArray; } - /* if (taosArrayGetSize(pColArray) == nCol) { - code = tdSTSRowNew(pColArray, pTSchema, ppRow); - if (code) goto _err; - } else { - *ppRow = NULL; - }*/ nextRowIterClose(&iter); - // taosArrayDestroy(pColArray); taosMemoryFreeClear(pTSchema); return code; _err: nextRowIterClose(&iter); - // taosArrayDestroy(pColArray); taosMemoryFreeClear(pTSchema); return code; } -// static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRow) { -// int32_t code = 0; -// SArray *pSkyline = NULL; - -// STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, uid, -1); -// int16_t nCol = pTSchema->numOfCols; -// SArray *pColArray = taosArrayInit(nCol, sizeof(SColVal)); - -// tb_uid_t suid = getTableSuidByUid(uid, pTsdb); - -// STbData *pMem = NULL; -// if (pTsdb->mem) { -// tsdbGetTbDataFromMemTable(pTsdb->mem, suid, uid, &pMem); -// } - -// STbData *pIMem = NULL; -// if (pTsdb->imem) { -// tsdbGetTbDataFromMemTable(pTsdb->imem, suid, uid, &pIMem); -// } - -// *ppRow = NULL; - -// pSkyline = taosArrayInit(32, sizeof(TSDBKEY)); - -// SDelIdx delIdx; - -// SDelFile *pDelFile = tsdbFSStateGetDelFile(pTsdb->pFS->cState); -// if (pDelFile) { -// SDelFReader *pDelFReader; - -// code = tsdbDelFReaderOpen(&pDelFReader, pDelFile, pTsdb, NULL); -// if (code) goto _err; - -// code = getTableDelIdx(pDelFReader, suid, uid, &delIdx); -// if (code) goto _err; - -// code = getTableDelSkyline(pMem, pIMem, pDelFReader, &delIdx, pSkyline); -// if (code) goto _err; - -// tsdbDelFReaderClose(&pDelFReader); -// } else { -// code = getTableDelSkyline(pMem, pIMem, NULL, NULL, pSkyline); -// if (code) goto _err; -// } - -// int64_t iSkyline = taosArrayGetSize(pSkyline) - 1; - -// SBlockIdx idx = {.suid = suid, .uid = uid}; - -// SFSNextRowIter fsState = {0}; -// fsState.state = SFSNEXTROW_FS; -// fsState.pTsdb = pTsdb; -// fsState.pBlockIdxExp = &idx; - -// SMemNextRowIter memState = {0}; -// SMemNextRowIter imemState = {0}; -// TSDBROW memRow, imemRow, fsRow; - -// TsdbNextRowState input[3] = {{&memRow, true, false, &memState, getNextRowFromMem, NULL}, -// {&imemRow, true, false, &imemState, getNextRowFromMem, NULL}, -// {&fsRow, false, true, &fsState, getNextRowFromFS, clearNextRowFromFS}}; - -// if (pMem) { -// memState.pMem = pMem; -// memState.state = SMEMNEXTROW_ENTER; -// input[0].stop = false; -// input[0].next = true; -// } -// if (pIMem) { -// imemState.pMem = pIMem; -// imemState.state = SMEMNEXTROW_ENTER; -// input[1].stop = false; -// input[1].next = true; -// } - -// int16_t nilColCount = nCol - 1; // count of null & none cols -// int iCol = 0; // index of first nil col index from left to right -// bool setICol = false; - -// do { -// for (int i = 0; i < 3; ++i) { -// if (input[i].next && !input[i].stop) { -// if (input[i].pRow == NULL) { -// code = input[i].nextRowFn(input[i].iter, &input[i].pRow); -// if (code) goto _err; - -// if (input[i].pRow == NULL) { -// input[i].stop = true; -// input[i].next = false; -// } -// } -// } -// } - -// if (input[0].stop && input[1].stop && input[2].stop) { -// break; -// } - -// // select maxpoint(s) from mem, imem, fs -// TSDBROW *max[3] = {0}; -// int iMax[3] = {-1, -1, -1}; -// int nMax = 0; -// TSKEY maxKey = TSKEY_MIN; - -// for (int i = 0; i < 3; ++i) { -// if (!input[i].stop && input[i].pRow != NULL) { -// TSDBKEY key = TSDBROW_KEY(input[i].pRow); - -// // merging & deduplicating on client side -// if (maxKey <= key.ts) { -// if (maxKey < key.ts) { -// nMax = 0; -// maxKey = key.ts; -// } - -// iMax[nMax] = i; -// max[nMax++] = input[i].pRow; -// } -// } -// } - -// // delete detection -// TSDBROW *merge[3] = {0}; -// int iMerge[3] = {-1, -1, -1}; -// int nMerge = 0; -// for (int i = 0; i < nMax; ++i) { -// TSDBKEY maxKey = TSDBROW_KEY(max[i]); - -// bool deleted = tsdbKeyDeleted(&maxKey, pSkyline, &iSkyline); -// if (!deleted) { -// iMerge[nMerge] = i; -// merge[nMerge++] = max[i]; -// } - -// input[iMax[i]].next = deleted; -// } - -// // merge if nMerge > 1 -// if (nMerge > 0) { -// *dup = false; - -// if (nMerge == 1) { -// code = tsRowFromTsdbRow(pTSchema, merge[nMerge - 1], ppRow); -// if (code) goto _err; -// } else { -// // merge 2 or 3 rows -// SRowMerger merger = {0}; - -// tRowMergerInit(&merger, merge[0], pTSchema); -// for (int i = 1; i < nMerge; ++i) { -// tRowMerge(&merger, merge[i]); -// } -// tRowMergerGetRow(&merger, ppRow); -// tRowMergerClear(&merger); -// } -// } - -// } while (1); - -// for (int i = 0; i < 3; ++i) { -// if (input[i].nextRowClearFn) { -// input[i].nextRowClearFn(input[i].iter); -// } -// } -// if (pSkyline) { -// taosArrayDestroy(pSkyline); -// } -// taosMemoryFreeClear(pTSchema); - -// return code; -// _err: -// for (int i = 0; i < 3; ++i) { -// if (input[i].nextRowClearFn) { -// input[i].nextRowClearFn(input[i].iter); -// } -// } -// if (pSkyline) { -// taosArrayDestroy(pSkyline); -// } -// taosMemoryFreeClear(pTSchema); -// tsdbError("vgId:%d merge last_row failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); -// return code; -// } - -// static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, STSRow **ppRow) { -// static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) { -// int32_t code = 0; -// SArray *pSkyline = NULL; -// STSRow *pRow = NULL; -// STSRow **ppRow = &pRow; - -// STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, uid, -1); -// int16_t nCol = pTSchema->numOfCols; -// // SArray *pColArray = taosArrayInit(nCol, sizeof(SColVal)); -// SArray *pColArray = taosArrayInit(nCol, sizeof(SLastCol)); - -// tb_uid_t suid = getTableSuidByUid(uid, pTsdb); - -// STbData *pMem = NULL; -// if (pTsdb->mem) { -// tsdbGetTbDataFromMemTable(pTsdb->mem, suid, uid, &pMem); -// } - -// STbData *pIMem = NULL; -// if (pTsdb->imem) { -// tsdbGetTbDataFromMemTable(pTsdb->imem, suid, uid, &pIMem); -// } - -// *ppLastArray = NULL; - -// pSkyline = taosArrayInit(32, sizeof(TSDBKEY)); - -// SDelIdx delIdx; - -// SDelFile *pDelFile = tsdbFSStateGetDelFile(pTsdb->pFS->cState); -// if (pDelFile) { -// SDelFReader *pDelFReader; - -// code = tsdbDelFReaderOpen(&pDelFReader, pDelFile, pTsdb, NULL); -// if (code) goto _err; - -// code = getTableDelIdx(pDelFReader, suid, uid, &delIdx); -// if (code) goto _err; - -// code = getTableDelSkyline(pMem, pIMem, pDelFReader, &delIdx, pSkyline); -// if (code) goto _err; - -// tsdbDelFReaderClose(&pDelFReader); -// } else { -// code = getTableDelSkyline(pMem, pIMem, NULL, NULL, pSkyline); -// if (code) goto _err; -// } - -// int64_t iSkyline = taosArrayGetSize(pSkyline) - 1; - -// SBlockIdx idx = {.suid = suid, .uid = uid}; - -// SFSNextRowIter fsState = {0}; -// fsState.state = SFSNEXTROW_FS; -// fsState.pTsdb = pTsdb; -// fsState.pBlockIdxExp = &idx; - -// SMemNextRowIter memState = {0}; -// SMemNextRowIter imemState = {0}; -// TSDBROW memRow, imemRow, fsRow; - -// TsdbNextRowState input[3] = {{&memRow, true, false, &memState, getNextRowFromMem, NULL}, -// {&imemRow, true, false, &imemState, getNextRowFromMem, NULL}, -// {&fsRow, false, true, &fsState, getNextRowFromFS, clearNextRowFromFS}}; - -// if (pMem) { -// memState.pMem = pMem; -// memState.state = SMEMNEXTROW_ENTER; -// input[0].stop = false; -// input[0].next = true; -// } -// if (pIMem) { -// imemState.pMem = pIMem; -// imemState.state = SMEMNEXTROW_ENTER; -// input[1].stop = false; -// input[1].next = true; -// } - -// int16_t nilColCount = nCol - 1; // count of null & none cols -// int iCol = 0; // index of first nil col index from left to right -// bool setICol = false; - -// do { -// for (int i = 0; i < 3; ++i) { -// if (input[i].next && !input[i].stop) { -// code = input[i].nextRowFn(input[i].iter, &input[i].pRow); -// if (code) goto _err; - -// if (input[i].pRow == NULL) { -// input[i].stop = true; -// input[i].next = false; -// } -// } -// } - -// if (input[0].stop && input[1].stop && input[2].stop) { -// break; -// } - -// // select maxpoint(s) from mem, imem, fs -// TSDBROW *max[3] = {0}; -// int iMax[3] = {-1, -1, -1}; -// int nMax = 0; -// TSKEY maxKey = TSKEY_MIN; - -// for (int i = 0; i < 3; ++i) { -// if (!input[i].stop && input[i].pRow != NULL) { -// TSDBKEY key = TSDBROW_KEY(input[i].pRow); - -// // merging & deduplicating on client side -// if (maxKey <= key.ts) { -// if (maxKey < key.ts) { -// nMax = 0; -// maxKey = key.ts; -// } - -// iMax[nMax] = i; -// max[nMax++] = input[i].pRow; -// } -// } -// } - -// // delete detection -// TSDBROW *merge[3] = {0}; -// int iMerge[3] = {-1, -1, -1}; -// int nMerge = 0; -// for (int i = 0; i < nMax; ++i) { -// TSDBKEY maxKey = TSDBROW_KEY(max[i]); - -// bool deleted = tsdbKeyDeleted(&maxKey, pSkyline, &iSkyline); -// if (!deleted) { -// iMerge[nMerge] = iMax[i]; -// merge[nMerge++] = max[i]; -// } - -// input[iMax[i]].next = deleted; -// } - -// // merge if nMerge > 1 -// if (nMerge > 0) { -// if (nMerge == 1) { -// code = tsRowFromTsdbRow(pTSchema, merge[nMerge - 1], ppRow); -// if (code) goto _err; -// } else { -// // merge 2 or 3 rows -// SRowMerger merger = {0}; - -// tRowMergerInit(&merger, merge[0], pTSchema); -// for (int i = 1; i < nMerge; ++i) { -// tRowMerge(&merger, merge[i]); -// } -// tRowMergerGetRow(&merger, ppRow); -// tRowMergerClear(&merger); -// } -// } else { -// /* *ppRow = NULL; */ -// /* return code; */ -// continue; -// } - -// if (iCol == 0) { -// STColumn *pTColumn = &pTSchema->columns[0]; -// SColVal *pColVal = &(SColVal){0}; - -// *pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, (SValue){.ts = maxKey}); - -// // if (taosArrayPush(pColArray, pColVal) == NULL) { -// if (taosArrayPush(pColArray, &(SLastCol){.ts = maxKey, .colVal = *pColVal}) == NULL) { -// code = TSDB_CODE_OUT_OF_MEMORY; -// goto _err; -// } - -// ++iCol; - -// setICol = false; -// for (int16_t i = iCol; i < nCol; ++i) { -// // tsdbRowGetColVal(*ppRow, pTSchema, i, pColVal); -// tTSRowGetVal(*ppRow, pTSchema, i, pColVal); -// // if (taosArrayPush(pColArray, pColVal) == NULL) { -// if (taosArrayPush(pColArray, &(SLastCol){.ts = maxKey, .colVal = *pColVal}) == NULL) { -// code = TSDB_CODE_OUT_OF_MEMORY; -// goto _err; -// } - -// if (pColVal->isNull || pColVal->isNone) { -// for (int j = 0; j < nMerge; ++j) { -// SColVal jColVal = {0}; -// tsdbRowGetColVal(merge[j], pTSchema, i, &jColVal); -// if (jColVal.isNull || jColVal.isNone) { -// input[iMerge[j]].next = true; -// } -// } -// if (!setICol) { -// iCol = i; -// setICol = true; -// } -// } else { -// --nilColCount; -// } -// } - -// if (*ppRow) { -// taosMemoryFreeClear(*ppRow); -// } - -// continue; -// } - -// setICol = false; -// for (int16_t i = iCol; i < nCol; ++i) { -// SColVal colVal = {0}; -// tTSRowGetVal(*ppRow, pTSchema, i, &colVal); -// TSKEY rowTs = (*ppRow)->ts; - -// // SColVal *tColVal = (SColVal *)taosArrayGet(pColArray, i); -// SLastCol *tTsVal = (SLastCol *)taosArrayGet(pColArray, i); -// SColVal *tColVal = &tTsVal->colVal; - -// if (!colVal.isNone && !colVal.isNull) { -// if (tColVal->isNull || tColVal->isNone) { -// // taosArraySet(pColArray, i, &colVal); -// taosArraySet(pColArray, i, &(SLastCol){.ts = rowTs, .colVal = colVal}); -// --nilColCount; -// } -// } else { -// if ((tColVal->isNull || tColVal->isNone) && !setICol) { -// iCol = i; -// setICol = true; - -// for (int j = 0; j < nMerge; ++j) { -// SColVal jColVal = {0}; -// tsdbRowGetColVal(merge[j], pTSchema, i, &jColVal); -// if (jColVal.isNull || jColVal.isNone) { -// input[iMerge[j]].next = true; -// } -// } -// } -// } -// } - -// if (*ppRow) { -// taosMemoryFreeClear(*ppRow); -// } -// } while (nilColCount > 0); - -// // if () new ts row from pColArray if non empty -// /* if (taosArrayGetSize(pColArray) == nCol) { */ -// /* code = tdSTSRowNew(pColArray, pTSchema, ppRow); */ -// /* if (code) goto _err; */ -// /* } */ -// /* taosArrayDestroy(pColArray); */ -// if (taosArrayGetSize(pColArray) <= 0) { -// *ppLastArray = NULL; -// taosArrayDestroy(pColArray); -// } else { -// *ppLastArray = pColArray; -// } -// if (*ppRow) { -// taosMemoryFreeClear(*ppRow); -// } - -// for (int i = 0; i < 3; ++i) { -// if (input[i].nextRowClearFn) { -// input[i].nextRowClearFn(input[i].iter); -// } -// } -// if (pSkyline) { -// taosArrayDestroy(pSkyline); -// } -// taosMemoryFreeClear(pTSchema); - -// return code; -// _err: -// taosArrayDestroy(pColArray); -// if (*ppRow) { -// taosMemoryFreeClear(*ppRow); -// } -// for (int i = 0; i < 3; ++i) { -// if (input[i].nextRowClearFn) { -// input[i].nextRowClearFn(input[i].iter); -// } -// } -// if (pSkyline) { -// taosArrayDestroy(pSkyline); -// } -// taosMemoryFreeClear(pTSchema); -// tsdbError("vgId:%d merge last_row failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); -// return code; -// } - int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUHandle **handle) { int32_t code = 0; char key[32] = {0}; int keyLen = 0; - // getTableCacheKey(uid, "lr", key, &keyLen); + // getTableCacheKeyS(uid, "lr", key, &keyLen); getTableCacheKey(uid, 0, key, &keyLen); LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen); if (h) { - //*ppRow = (STSRow *)taosLRUCacheValue(pCache, h); } else { STSRow *pRow = NULL; bool dup = false; // which is always false for now - code = mergeLastRow2(uid, pTsdb, &dup, &pRow); + code = mergeLastRow(uid, pTsdb, &dup, &pRow); // if table's empty or error, return code of -1 if (code < 0 || pRow == NULL) { if (!dup && pRow) { @@ -1680,9 +1127,7 @@ int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUH code = -1; } - // tsdbCacheInsertLastrow(pCache, pTsdb, uid, pRow, dup); h = taosLRUCacheLookup(pCache, key, keyLen); - //*ppRow = (STSRow *)taosLRUCacheValue(pCache, h); } *handle = h; @@ -1719,18 +1164,13 @@ int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUHand char key[32] = {0}; int keyLen = 0; - // getTableCacheKey(uid, "l", key, &keyLen); + // getTableCacheKeyS(uid, "l", key, &keyLen); getTableCacheKey(uid, 1, key, &keyLen); LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen); if (h) { - //*ppRow = (STSRow *)taosLRUCacheValue(pCache, h); - } else { - // STSRow *pRow = NULL; - // code = mergeLast(uid, pTsdb, &pRow); SArray *pLastArray = NULL; - // code = mergeLast(uid, pTsdb, &pLastArray); - code = mergeLast2(uid, pTsdb, &pLastArray); + code = mergeLast(uid, pTsdb, &pLastArray); // if table's empty or error, return code of -1 // if (code < 0 || pRow == NULL) { if (code < 0 || pLastArray == NULL) { @@ -1746,7 +1186,6 @@ int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUHand } h = taosLRUCacheLookup(pCache, key, keyLen); - //*ppRow = (STSRow *)taosLRUCacheValue(pCache, h); } *handle = h; diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c index 4323fa0aff0955c9b176dfeddac37f5c21cdcfd5..a0e5071685bb0870c92cf7be968772ee304f7234 100644 --- a/source/dnode/vnode/src/vnd/vnodeSync.c +++ b/source/dnode/vnode/src/vnd/vnodeSync.c @@ -281,8 +281,8 @@ void vnodeApplyWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { for (int32_t i = 0; i < numOfMsgs; ++i) { if (taosGetQitem(qall, (void **)&pMsg) == 0) continue; const STraceId *trace = &pMsg->info.traceId; - vGTrace("vgId:%d, msg:%p get from vnode-apply queue, type:%s handle:%p", vgId, pMsg, TMSG_INFO(pMsg->msgType), - pMsg->info.handle); + vGInfo("vgId:%d, msg:%p get from vnode-apply queue, type:%s handle:%p index:%ld", vgId, pMsg, + TMSG_INFO(pMsg->msgType), pMsg->info.handle, pMsg->info.conn.applyIndex); SRpcMsg rsp = {.code = pMsg->code, .info = pMsg->info}; if (rsp.code == 0) { @@ -503,9 +503,6 @@ static void vnodeSyncReconfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReCon static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { if (cbMeta.isWeak == 0) { SVnode *pVnode = pFsm->data; - vTrace("vgId:%d, commit-cb is excuted, fsm:%p, index:%" PRId64 ", isWeak:%d, code:%d, state:%d %s, msgtype:%d %s", - syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, - syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType)); if (cbMeta.code == 0) { SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen}; @@ -514,11 +511,17 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info); rpcMsg.info.conn.applyIndex = cbMeta.index; rpcMsg.info.conn.applyTerm = cbMeta.term; + + vInfo("vgId:%d, commit-cb is excuted, fsm:%p, index:%" PRId64 ", term:%" PRIu64 ", msg-index:%" PRId64 + ", isWeak:%d, code:%d, state:%d %s, msgtype:%d %s", + syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.term, rpcMsg.info.conn.applyIndex, cbMeta.isWeak, + cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType)); + tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg); } else { SRpcMsg rsp = {.code = cbMeta.code, .info = pMsg->info}; - vError("vgId:%d, sync commit error, msgtype:%d,%s, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync), - pMsg->msgType, TMSG_INFO(pMsg->msgType), cbMeta.code, tstrerror(cbMeta.code)); + vError("vgId:%d, sync commit error, msgtype:%d,%s, index:%ld, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync), + pMsg->msgType, TMSG_INFO(pMsg->msgType), cbMeta.index, cbMeta.code, tstrerror(cbMeta.code)); if (rsp.info.handle != NULL) { tmsgSendRsp(&rsp); } diff --git a/source/libs/index/src/indexFilter.c b/source/libs/index/src/indexFilter.c index 7fc41b8dff3a6344186ce0bf82c83762b15d8a81..e1c8ac0204f77f6e4756cd05fba36141b911290f 100644 --- a/source/libs/index/src/indexFilter.c +++ b/source/libs/index/src/indexFilter.c @@ -579,11 +579,13 @@ static int32_t sifExecLogic(SLogicConditionNode *node, SIFCtx *ctx, SIFParam *ou if (ctx->noExec == false) { for (int32_t m = 0; m < node->pParameterList->length; m++) { - // add impl later if (node->condType == LOGIC_COND_TYPE_AND) { taosArrayAddAll(output->result, params[m].result); + // taosArrayDestroy(params[m].result); + // params[m].result = NULL; } else if (node->condType == LOGIC_COND_TYPE_OR) { taosArrayAddAll(output->result, params[m].result); + // params[m].result = NULL; } else if (node->condType == LOGIC_COND_TYPE_NOT) { // taosArrayAddAll(output->result, params[m].result); } @@ -593,6 +595,8 @@ static int32_t sifExecLogic(SLogicConditionNode *node, SIFCtx *ctx, SIFParam *ou } else { for (int32_t m = 0; m < node->pParameterList->length; m++) { output->status = sifMergeCond(node->condType, output->status, params[m].status); + taosArrayDestroy(params[m].result); + params[m].result = NULL; } } _return: @@ -607,6 +611,7 @@ static EDealRes sifWalkFunction(SNode *pNode, void *context) { SIFCtx *ctx = context; ctx->code = sifExecFunction(node, ctx, &output); if (ctx->code != TSDB_CODE_SUCCESS) { + sifFreeParam(&output); return DEAL_RES_ERROR; } @@ -624,6 +629,7 @@ static EDealRes sifWalkLogic(SNode *pNode, void *context) { SIFCtx *ctx = context; ctx->code = sifExecLogic(node, ctx, &output); if (ctx->code) { + sifFreeParam(&output); return DEAL_RES_ERROR; } @@ -640,6 +646,7 @@ static EDealRes sifWalkOper(SNode *pNode, void *context) { SIFCtx *ctx = context; ctx->code = sifExecOper(node, ctx, &output); if (ctx->code) { + sifFreeParam(&output); return DEAL_RES_ERROR; } if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) { @@ -698,7 +705,11 @@ static int32_t sifCalculate(SNode *pNode, SIFParam *pDst) { } nodesWalkExprPostOrder(pNode, sifCalcWalker, &ctx); - SIF_ERR_RET(ctx.code); + + if (ctx.code != 0) { + sifFreeRes(ctx.pRes); + return ctx.code; + } if (pDst) { SIFParam *res = (SIFParam *)taosHashGet(ctx.pRes, (void *)&pNode, POINTER_BYTES); @@ -714,8 +725,7 @@ static int32_t sifCalculate(SNode *pNode, SIFParam *pDst) { taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES); } sifFreeRes(ctx.pRes); - - SIF_RET(code); + return code; } static int32_t sifGetFltHint(SNode *pNode, SIdxFltStatus *status) { @@ -732,8 +742,10 @@ static int32_t sifGetFltHint(SNode *pNode, SIdxFltStatus *status) { } nodesWalkExprPostOrder(pNode, sifCalcWalker, &ctx); - - SIF_ERR_RET(ctx.code); + if (ctx.code != 0) { + sifFreeRes(ctx.pRes); + return ctx.code; + } SIFParam *res = (SIFParam *)taosHashGet(ctx.pRes, (void *)&pNode, POINTER_BYTES); if (res == NULL) { @@ -745,8 +757,7 @@ static int32_t sifGetFltHint(SNode *pNode, SIdxFltStatus *status) { sifFreeParam(res); taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES); taosHashCleanup(ctx.pRes); - - SIF_RET(code); + return code; } int32_t doFilterTag(SNode *pFilterNode, SIndexMetaArg *metaArg, SArray *result, SIdxFltStatus *status) { @@ -760,7 +771,11 @@ int32_t doFilterTag(SNode *pFilterNode, SIndexMetaArg *metaArg, SArray *result, SArray *output = taosArrayInit(8, sizeof(uint64_t)); SIFParam param = {.arg = *metaArg, .result = output}; - SIF_ERR_RET(sifCalculate((SNode *)pFilterNode, ¶m)); + int32_t code = sifCalculate((SNode *)pFilterNode, ¶m); + if (code != 0) { + sifFreeParam(¶m); + return code; + } taosArrayAddAll(result, param.result); sifFreeParam(¶m); diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 064c110a9f871582426ec1deb1e666a3760253b6..efbe110f6f224b2740281dadf420b1e551fb4c2c 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -1,5 +1,4 @@ /** Copyright (c) 2019 TAOS Data, Inc. - * * This program is free software: you can use, redistribute, and/or modify * it under the terms of the GNU Affero General Public License, version 3 @@ -16,6 +15,10 @@ #ifdef USE_UV #include "transComm.h" +typedef struct SConnList { + queue conn; +} SConnList; + typedef struct SCliConn { T_REF_DECLARE() uv_connect_t connReq; @@ -26,7 +29,9 @@ typedef struct SCliConn { SConnBuffer readBuf; STransQueue cliMsgs; - queue q; + + queue q; + SConnList* list; STransCtx ctx; bool broken; // link broken or not @@ -56,13 +61,14 @@ typedef struct SCliMsg { } SCliMsg; typedef struct SCliThrd { - TdThread thread; // tid - int64_t pid; // pid - uv_loop_t* loop; - SAsyncPool* asyncPool; - uv_idle_t* idle; - uv_timer_t timer; - void* pool; // conn pool + TdThread thread; // tid + int64_t pid; // pid + uv_loop_t* loop; + SAsyncPool* asyncPool; + uv_idle_t* idle; + uv_prepare_t* prepare; + uv_timer_t timer; + void* pool; // conn pool // msg queue queue msg; @@ -86,10 +92,6 @@ typedef struct SCliObj { SCliThrd** pThreadObj; } SCliObj; -typedef struct SConnList { - queue conn; -} SConnList; - // conn pool // add expire timeout and capacity limit static void* createConnPool(int size); @@ -101,7 +103,7 @@ static void doCloseIdleConn(void* param); static int sockDebugInfo(struct sockaddr* sockname, char* dst) { struct sockaddr_in addr = *(struct sockaddr_in*)sockname; - char buf[20] = {0}; + char buf[16] = {0}; int r = uv_ip4_name(&addr, (char*)buf, sizeof(buf)); sprintf(dst, "%s:%d", buf, ntohs(addr.sin_port)); return r; @@ -118,6 +120,9 @@ static void cliSendCb(uv_write_t* req, int status); static void cliConnCb(uv_connect_t* req, int status); static void cliAsyncCb(uv_async_t* handle); static void cliIdleCb(uv_idle_t* handle); +static void cliPrepareCb(uv_prepare_t* handle); + +static int32_t allocConnRef(SCliConn* conn, bool update); static int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg); @@ -198,7 +203,7 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) { pThrd = (SCliThrd*)(exh)->pThrd; \ } \ } while (0) -#define CONN_PERSIST_TIME(para) ((para) == 0 ? 3 * 1000 : (para)) +#define CONN_PERSIST_TIME(para) ((para) <= 90000 ? 90000 : (para)) #define CONN_GET_HOST_THREAD(conn) (conn ? ((SCliConn*)conn)->hostThrd : NULL) #define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrd*)(conn)->hostThrd)->pTransInst))->label) #define CONN_SHOULD_RELEASE(conn, head) \ @@ -499,9 +504,8 @@ void* destroyConnPool(void* pool) { } static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) { - char key[128] = {0}; + char key[32] = {0}; CONN_CONSTRUCT_HASH_KEY(key, ip, port); - SHashObj* pPool = pool; SConnList* plist = taosHashGet(pPool, key, strlen(key)); if (plist == NULL) { @@ -519,13 +523,44 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) { conn->status = ConnNormal; QUEUE_REMOVE(&conn->q); QUEUE_INIT(&conn->q); - assert(h == &conn->q); transDQCancel(((SCliThrd*)conn->hostThrd)->timeoutQueue, conn->task); conn->task = NULL; return conn; } +static void addConnToPool(void* pool, SCliConn* conn) { + if (conn->status == ConnInPool) { + return; + } + SCliThrd* thrd = conn->hostThrd; + CONN_HANDLE_THREAD_QUIT(thrd); + + allocConnRef(conn, true); + + STrans* pTransInst = thrd->pTransInst; + cliReleaseUnfinishedMsg(conn); + transQueueClear(&conn->cliMsgs); + transCtxCleanup(&conn->ctx); + conn->status = ConnInPool; + + if (conn->list == NULL) { + char key[32] = {0}; + CONN_CONSTRUCT_HASH_KEY(key, conn->ip, conn->port); + tTrace("%s conn %p added to conn pool, read buf cap:%d", CONN_GET_INST_LABEL(conn), conn, conn->readBuf.cap); + conn->list = taosHashGet((SHashObj*)pool, key, strlen(key)); + } + assert(conn->list != NULL); + QUEUE_INIT(&conn->q); + QUEUE_PUSH(&conn->list->conn, &conn->q); + + assert(!QUEUE_IS_EMPTY(&conn->list->conn)); + + STaskArg* arg = taosMemoryCalloc(1, sizeof(STaskArg)); + arg->param1 = conn; + arg->param2 = thrd; + conn->task = transDQSched(thrd->timeoutQueue, doCloseIdleConn, arg, CONN_PERSIST_TIME(pTransInst->idleTime)); +} static int32_t allocConnRef(SCliConn* conn, bool update) { if (update) { transRemoveExHandle(transGetRefMgt(), conn->refId); @@ -556,38 +591,6 @@ static int32_t specifyConnRef(SCliConn* conn, bool update, int64_t handle) { return 0; } -static void addConnToPool(void* pool, SCliConn* conn) { - if (conn->status == ConnInPool) { - return; - } - SCliThrd* thrd = conn->hostThrd; - CONN_HANDLE_THREAD_QUIT(thrd); - - allocConnRef(conn, true); - - STrans* pTransInst = thrd->pTransInst; - cliReleaseUnfinishedMsg(conn); - transQueueClear(&conn->cliMsgs); - transCtxCleanup(&conn->ctx); - conn->status = ConnInPool; - - char key[128] = {0}; - CONN_CONSTRUCT_HASH_KEY(key, conn->ip, conn->port); - tTrace("%s conn %p added to conn pool, read buf cap:%d", CONN_GET_INST_LABEL(conn), conn, conn->readBuf.cap); - - SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key)); - // list already create before - assert(plist != NULL); - QUEUE_INIT(&conn->q); - QUEUE_PUSH(&plist->conn, &conn->q); - - assert(!QUEUE_IS_EMPTY(&plist->conn)); - - STaskArg* arg = taosMemoryCalloc(1, sizeof(STaskArg)); - arg->param1 = conn; - arg->param2 = thrd; - conn->task = transDQSched(thrd->timeoutQueue, doCloseIdleConn, arg, CONN_PERSIST_TIME(pTransInst->idleTime)); -} static void cliAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { SCliConn* conn = handle->data; SConnBuffer* pBuf = &conn->readBuf; @@ -602,11 +605,9 @@ static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { SConnBuffer* pBuf = &conn->readBuf; if (nread > 0) { pBuf->len += nread; - if (transReadComplete(pBuf)) { + while (transReadComplete(pBuf)) { tTrace("%s conn %p read complete", CONN_GET_INST_LABEL(conn), conn); cliHandleResp(conn); - } else { - tTrace("%s conn %p read partial packet, continue to read", CONN_GET_INST_LABEL(conn), conn); } return; } @@ -967,6 +968,62 @@ static void cliAsyncCb(uv_async_t* handle) { static void cliIdleCb(uv_idle_t* handle) { SCliThrd* thrd = handle->data; tTrace("do idle work"); + + SAsyncPool* pool = thrd->asyncPool; + for (int i = 0; i < pool->nAsync; i++) { + uv_async_t* async = &(pool->asyncs[i]); + SAsyncItem* item = async->data; + + queue wq; + taosThreadMutexLock(&item->mtx); + QUEUE_MOVE(&item->qmsg, &wq); + taosThreadMutexUnlock(&item->mtx); + + int count = 0; + while (!QUEUE_IS_EMPTY(&wq)) { + queue* h = QUEUE_HEAD(&wq); + QUEUE_REMOVE(h); + + SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q); + if (pMsg == NULL) { + continue; + } + (*cliAsyncHandle[pMsg->type])(pMsg, thrd); + count++; + } + } + tTrace("prepare work end"); + if (thrd->stopMsg != NULL) cliHandleQuit(thrd->stopMsg, thrd); +} +static void cliPrepareCb(uv_prepare_t* handle) { + SCliThrd* thrd = handle->data; + tTrace("prepare work start"); + + SAsyncPool* pool = thrd->asyncPool; + for (int i = 0; i < pool->nAsync; i++) { + uv_async_t* async = &(pool->asyncs[i]); + SAsyncItem* item = async->data; + + queue wq; + taosThreadMutexLock(&item->mtx); + QUEUE_MOVE(&item->qmsg, &wq); + taosThreadMutexUnlock(&item->mtx); + + int count = 0; + while (!QUEUE_IS_EMPTY(&wq)) { + queue* h = QUEUE_HEAD(&wq); + QUEUE_REMOVE(h); + + SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q); + if (pMsg == NULL) { + continue; + } + (*cliAsyncHandle[pMsg->type])(pMsg, thrd); + count++; + } + } + tTrace("prepare work end"); + if (thrd->stopMsg != NULL) cliHandleQuit(thrd->stopMsg, thrd); } static void* cliWorkThread(void* arg) { @@ -1033,7 +1090,12 @@ static SCliThrd* createThrdObj() { // pThrd->idle = taosMemoryCalloc(1, sizeof(uv_idle_t)); // uv_idle_init(pThrd->loop, pThrd->idle); // pThrd->idle->data = pThrd; - // uv_idle_start(pThrd->idle, cliIdleCb); + // uv_idle_start(pThrd->idle, cliIdleCb); + + pThrd->prepare = taosMemoryCalloc(1, sizeof(uv_prepare_t)); + uv_prepare_init(pThrd->loop, pThrd->prepare); + pThrd->prepare->data = pThrd; + uv_prepare_start(pThrd->prepare, cliPrepareCb); pThrd->pool = createConnPool(4); transDQCreate(pThrd->loop, &pThrd->delayQueue); @@ -1058,6 +1120,7 @@ static void destroyThrdObj(SCliThrd* pThrd) { transDQDestroy(pThrd->timeoutQueue, NULL); taosMemoryFree(pThrd->idle); + taosMemoryFree(pThrd->prepare); taosMemoryFree(pThrd->loop); taosMemoryFree(pThrd); } diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index c99effb26f4628986e9c3be266219ff371e8ff93..8cf525a506fe856876f2e8577e6f97dedbdb8d26 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -120,8 +120,9 @@ int transInitBuffer(SConnBuffer* buf) { buf->total = 0; return 0; } -int transDestroyBuffer(SConnBuffer* buf) { - taosMemoryFree(buf->buf); +int transDestroyBuffer(SConnBuffer* p) { + taosMemoryFree(p->buf); + p->buf = NULL; return 0; } diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index a97e0b53c1c5c4b443ddabe4061885bc64ee824d..4b579a1f9527bde98f9ac4d21723ed9d17e965d6 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -73,6 +73,7 @@ typedef struct SWorkThrd { uv_os_fd_t fd; uv_loop_t* loop; SAsyncPool* asyncPool; + uv_prepare_t* prepare; queue msg; TdThreadMutex msgMtx; @@ -112,6 +113,7 @@ static void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) static void uvWorkerAsyncCb(uv_async_t* handle); static void uvAcceptAsyncCb(uv_async_t* handle); static void uvShutDownCb(uv_shutdown_t* req, int status); +static void uvPrepareCb(uv_prepare_t* handle); /* * time-consuming task throwed into BG work thread @@ -238,8 +240,6 @@ static void uvHandleReq(SSvrConn* pConn) { transMsg.msgType = pHead->msgType; transMsg.code = pHead->code; - // transClearBuffer(&pConn->readBuf); - pConn->inType = pHead->msgType; if (pConn->status == ConnNormal) { if (pHead->persist == 1) { @@ -546,6 +546,52 @@ static void uvShutDownCb(uv_shutdown_t* req, int status) { uv_close((uv_handle_t*)req->handle, uvDestroyConn); taosMemoryFree(req); } +static void uvPrepareCb(uv_prepare_t* handle) { + // prepare callback + SWorkThrd* pThrd = handle->data; + SAsyncPool* pool = pThrd->asyncPool; + + for (int i = 0; i < pool->nAsync; i++) { + uv_async_t* async = &(pool->asyncs[i]); + SAsyncItem* item = async->data; + + queue wq; + taosThreadMutexLock(&item->mtx); + QUEUE_MOVE(&item->qmsg, &wq); + taosThreadMutexUnlock(&item->mtx); + + while (!QUEUE_IS_EMPTY(&wq)) { + queue* head = QUEUE_HEAD(&wq); + QUEUE_REMOVE(head); + + SSvrMsg* msg = QUEUE_DATA(head, SSvrMsg, q); + if (msg == NULL) { + tError("unexcept occurred, continue"); + continue; + } + // release handle to rpc init + if (msg->type == Quit) { + (*transAsyncHandle[msg->type])(msg, pThrd); + continue; + } else { + STransMsg transMsg = msg->msg; + + SExHandle* exh1 = transMsg.info.handle; + int64_t refId = transMsg.info.refId; + SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), refId); + if (exh2 == NULL || exh1 != exh2) { + tTrace("handle except msg %p, ignore it", exh1); + transReleaseExHandle(transGetRefMgt(), refId); + destroySmsg(msg); + continue; + } + msg->pConn = exh1->handle; + transReleaseExHandle(transGetRefMgt(), refId); + (*transAsyncHandle[msg->type])(msg, pThrd); + } + } + } +} static void uvWorkDoTask(uv_work_t* req) { // doing time-consumeing task @@ -695,13 +741,17 @@ static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) { } uv_pipe_init(pThrd->loop, pThrd->pipe, 1); - // int r = uv_pipe_open(pThrd->pipe, pThrd->fd); pThrd->pipe->data = pThrd; QUEUE_INIT(&pThrd->msg); taosThreadMutexInit(&pThrd->msgMtx, NULL); + pThrd->prepare = taosMemoryCalloc(1, sizeof(uv_prepare_t)); + uv_prepare_init(pThrd->loop, pThrd->prepare); + uv_prepare_start(pThrd->prepare, uvPrepareCb); + pThrd->prepare->data = pThrd; + // conn set QUEUE_INIT(&pThrd->conn); @@ -986,6 +1036,7 @@ void destroyWorkThrd(SWorkThrd* pThrd) { SRV_RELEASE_UV(pThrd->loop); TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSvrMsg, destroySmsg); transAsyncPoolDestroy(pThrd->asyncPool); + taosMemoryFree(pThrd->prepare); taosMemoryFree(pThrd->loop); taosMemoryFree(pThrd); } diff --git a/source/libs/wal/src/walMeta.c b/source/libs/wal/src/walMeta.c index a5fd3fca35c01b84f0b2c5b7cc4fe3b7bbf76662..a8da6809100fa5789e5b7e57e051631257782e1e 100644 --- a/source/libs/wal/src/walMeta.c +++ b/source/libs/wal/src/walMeta.c @@ -139,7 +139,7 @@ int walCheckAndRepairMeta(SWal* pWal) { const char* idxPattern = "^[0-9]+.idx$"; regex_t logRegPattern; regex_t idxRegPattern; - SArray* pLogInfoArray = taosArrayInit(8, sizeof(SWalFileInfo)); + SArray* actualLog = taosArrayInit(8, sizeof(SWalFileInfo)); regcomp(&logRegPattern, logPattern, REG_EXTENDED); regcomp(&idxRegPattern, idxPattern, REG_EXTENDED); @@ -159,7 +159,7 @@ int walCheckAndRepairMeta(SWal* pWal) { SWalFileInfo fileInfo; memset(&fileInfo, -1, sizeof(SWalFileInfo)); sscanf(name, "%" PRId64 ".log", &fileInfo.firstVer); - taosArrayPush(pLogInfoArray, &fileInfo); + taosArrayPush(actualLog, &fileInfo); } } @@ -167,10 +167,10 @@ int walCheckAndRepairMeta(SWal* pWal) { regfree(&logRegPattern); regfree(&idxRegPattern); - taosArraySort(pLogInfoArray, compareWalFileInfo); + taosArraySort(actualLog, compareWalFileInfo); int metaFileNum = taosArrayGetSize(pWal->fileInfoSet); - int actualFileNum = taosArrayGetSize(pLogInfoArray); + int actualFileNum = taosArrayGetSize(actualLog); #if 0 for (int32_t fileNo = actualFileNum - 1; fileNo >= 0; fileNo--) { @@ -196,11 +196,11 @@ int walCheckAndRepairMeta(SWal* pWal) { taosArrayPopFrontBatch(pWal->fileInfoSet, metaFileNum - actualFileNum); } else if (metaFileNum < actualFileNum) { for (int i = metaFileNum; i < actualFileNum; i++) { - SWalFileInfo* pFileInfo = taosArrayGet(pLogInfoArray, i); + SWalFileInfo* pFileInfo = taosArrayGet(actualLog, i); taosArrayPush(pWal->fileInfoSet, pFileInfo); } } - taosArrayDestroy(pLogInfoArray); + taosArrayDestroy(actualLog); pWal->writeCur = actualFileNum - 1; if (actualFileNum > 0) { @@ -221,7 +221,7 @@ int walCheckAndRepairMeta(SWal* pWal) { int code = walSaveMeta(pWal); if (code < 0) { - taosArrayDestroy(pLogInfoArray); + taosArrayDestroy(actualLog); return -1; } } diff --git a/source/libs/wal/src/walRead.c b/source/libs/wal/src/walRead.c index ac62b7d98dfeec5e7df073ef165a79d1ad95b7f6..787c9af31703df50a4b91589e20f5f9373d71c56 100644 --- a/source/libs/wal/src/walRead.c +++ b/source/libs/wal/src/walRead.c @@ -423,37 +423,38 @@ int32_t walFetchBody(SWalReader *pRead, SWalCkHead **ppHead) { return 0; } -int32_t walReadVer(SWalReader *pRead, int64_t ver) { - wDebug("vgId:%d wal start to read ver %ld", pRead->pWal->cfg.vgId, ver); +int32_t walReadVer(SWalReader *pReader, int64_t ver) { + wDebug("vgId:%d wal start to read ver %ld", pReader->pWal->cfg.vgId, ver); int64_t contLen; + int32_t code; bool seeked = false; - if (pRead->pWal->vers.firstVer == -1) { + if (pReader->pWal->vers.firstVer == -1) { terrno = TSDB_CODE_WAL_LOG_NOT_EXIST; return -1; } - if (ver > pRead->pWal->vers.lastVer || ver < pRead->pWal->vers.firstVer) { - wDebug("vgId:%d, invalid index:%" PRId64 ", first index:%" PRId64 ", last index:%" PRId64, pRead->pWal->cfg.vgId, - ver, pRead->pWal->vers.firstVer, pRead->pWal->vers.lastVer); + if (ver > pReader->pWal->vers.lastVer || ver < pReader->pWal->vers.firstVer) { + wDebug("vgId:%d, invalid index:%" PRId64 ", first index:%" PRId64 ", last index:%" PRId64, pReader->pWal->cfg.vgId, + ver, pReader->pWal->vers.firstVer, pReader->pWal->vers.lastVer); terrno = TSDB_CODE_WAL_LOG_NOT_EXIST; return -1; } - if (pRead->curInvalid || pRead->curVersion != ver) { - if (walReadSeekVer(pRead, ver) < 0) { - wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since %s", pRead->pWal->cfg.vgId, ver, terrstr()); + if (pReader->curInvalid || pReader->curVersion != ver) { + if (walReadSeekVer(pReader, ver) < 0) { + wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since %s", pReader->pWal->cfg.vgId, ver, terrstr()); return -1; } seeked = true; } while (1) { - contLen = taosReadFile(pRead->pLogFile, pRead->pHead, sizeof(SWalCkHead)); + contLen = taosReadFile(pReader->pLogFile, pReader->pHead, sizeof(SWalCkHead)); if (contLen == sizeof(SWalCkHead)) { break; } else if (contLen == 0 && !seeked) { - walReadSeekVerImpl(pRead, ver); + walReadSeekVerImpl(pReader, ver); seeked = true; continue; } else { @@ -467,26 +468,26 @@ int32_t walReadVer(SWalReader *pRead, int64_t ver) { } } - contLen = walValidHeadCksum(pRead->pHead); - if (contLen != 0) { - wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since head checksum not passed", pRead->pWal->cfg.vgId, + code = walValidHeadCksum(pReader->pHead); + if (code != 0) { + wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since head checksum not passed", pReader->pWal->cfg.vgId, ver); terrno = TSDB_CODE_WAL_FILE_CORRUPTED; return -1; } - if (pRead->capacity < pRead->pHead->head.bodyLen) { - void *ptr = taosMemoryRealloc(pRead->pHead, sizeof(SWalCkHead) + pRead->pHead->head.bodyLen); + if (pReader->capacity < pReader->pHead->head.bodyLen) { + void *ptr = taosMemoryRealloc(pReader->pHead, sizeof(SWalCkHead) + pReader->pHead->head.bodyLen); if (ptr == NULL) { terrno = TSDB_CODE_WAL_OUT_OF_MEMORY; return -1; } - pRead->pHead = ptr; - pRead->capacity = pRead->pHead->head.bodyLen; + pReader->pHead = ptr; + pReader->capacity = pReader->pHead->head.bodyLen; } - if ((contLen = taosReadFile(pRead->pLogFile, pRead->pHead->head.body, pRead->pHead->head.bodyLen)) != - pRead->pHead->head.bodyLen) { + if ((contLen = taosReadFile(pReader->pLogFile, pReader->pHead->head.body, pReader->pHead->head.bodyLen)) != + pReader->pHead->head.bodyLen) { if (contLen < 0) terrno = TAOS_SYSTEM_ERROR(errno); else { @@ -496,25 +497,28 @@ int32_t walReadVer(SWalReader *pRead, int64_t ver) { return -1; } - if (pRead->pHead->head.version != ver) { - wError("vgId:%d, unexpected wal log, index:%" PRId64 ", read request index:%" PRId64, pRead->pWal->cfg.vgId, - pRead->pHead->head.version, ver); - pRead->curInvalid = 1; + if (pReader->pHead->head.version != ver) { + wError("vgId:%d, unexpected wal log, index:%" PRId64 ", read request index:%" PRId64, pReader->pWal->cfg.vgId, + pReader->pHead->head.version, ver); + pReader->curInvalid = 1; terrno = TSDB_CODE_WAL_FILE_CORRUPTED; ASSERT(0); return -1; } - contLen = walValidBodyCksum(pRead->pHead); - if (contLen != 0) { - wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since body checksum not passed", pRead->pWal->cfg.vgId, + code = walValidBodyCksum(pReader->pHead); + if (code != 0) { + wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since body checksum not passed", pReader->pWal->cfg.vgId, ver); - pRead->curInvalid = 1; + uint32_t readCkSum = walCalcBodyCksum(pReader->pHead->head.body, pReader->pHead->head.bodyLen); + uint32_t logCkSum = pReader->pHead->cksumBody; + wError("checksum written into log: %u, checksum calculated: %u", logCkSum, readCkSum); + pReader->curInvalid = 1; terrno = TSDB_CODE_WAL_FILE_CORRUPTED; ASSERT(0); return -1; } - pRead->curVersion++; + pReader->curVersion++; return 0; } diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 491e5b0e08ff190ed9b042170e942738a22e5f36..4eadc92f705da24df20555ffef1bf82d7fca9858 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -289,18 +289,25 @@ int32_t walEndSnapshot(SWal *pWal) { newTotSize -= iter->fileSize; } } - char fnameStr[WAL_FILE_LEN]; + int32_t actualDelete = 0; + char fnameStr[WAL_FILE_LEN]; // remove file for (int i = 0; i < deleteCnt; i++) { pInfo = taosArrayGet(pWal->fileInfoSet, i); walBuildLogName(pWal, pInfo->firstVer, fnameStr); - taosRemoveFile(fnameStr); + if (taosRemoveFile(fnameStr) < 0) { + goto UPDATE_META; + } walBuildIdxName(pWal, pInfo->firstVer, fnameStr); - taosRemoveFile(fnameStr); + if (taosRemoveFile(fnameStr) < 0) { + ASSERT(0); + } + actualDelete++; } + UPDATE_META: // make new array, remove files - taosArrayPopFrontBatch(pWal->fileInfoSet, deleteCnt); + taosArrayPopFrontBatch(pWal->fileInfoSet, actualDelete); if (taosArrayGetSize(pWal->fileInfoSet) == 0) { pWal->writeCur = -1; pWal->vers.firstVer = -1; diff --git a/tests/script/jenkins/basic.txt b/tests/script/jenkins/basic.txt index 4a6a1ca0b8d0a889a41dcad534f368cdbd061047..6de5a9ab98cc4ac5b6f20deae355a7021345dca6 100644 --- a/tests/script/jenkins/basic.txt +++ b/tests/script/jenkins/basic.txt @@ -166,7 +166,7 @@ # ---- query ---- ./test.sh -f tsim/query/charScalarFunction.sim -# ./test.sh -f tsim/query/explain.sim +./test.sh -f tsim/query/explain.sim ./test.sh -f tsim/query/interval-offset.sim ./test.sh -f tsim/query/interval.sim ./test.sh -f tsim/query/scalarFunction.sim