Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 7fe743d0 编写于 作者: H Hongze Cheng
浏览文件
操作

refact code for further change

上级 f963f5bf
隐藏空白更改
内联 并排
Showing with 421 addition and 422 deletion
source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
浏览文件 @ 7fe743d0
...@@ -14,382 +14,136 @@ ...@@ -14,382 +14,136 @@
*/ */
#include "tsdb.h" #include "tsdb.h"
// =============== PAGE-WISE FILE ===============
typedef struct {
TdFilePtr pFD;
int32_t szPage;
int32_t nBuf;
uint8_t *pBuf;
int64_t pgno;
} STsdbFD;
// SDelFWriter ==================================================== int32_t tsdbOpenFile(const char *path, int32_t opt, STsdbFD *pFD) {
int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb) {
int32_t code = 0;
char fname[TSDB_FILENAME_LEN];
char hdr[TSDB_FHDR_SIZE] = {0};
SDelFWriter *pDelFWriter;
int64_t n;
// alloc
pDelFWriter = (SDelFWriter *)taosMemoryCalloc(1, sizeof(*pDelFWriter));
if (pDelFWriter == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pDelFWriter->pTsdb = pTsdb;
pDelFWriter->fDel = *pFile;
tsdbDelFileName(pTsdb, pFile, fname);
pDelFWriter->pWriteH = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE);
if (pDelFWriter->pWriteH == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// update header
n = taosWriteFile(pDelFWriter->pWriteH, &hdr, TSDB_FHDR_SIZE);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
pDelFWriter->fDel.size = TSDB_FHDR_SIZE;
pDelFWriter->fDel.offset = 0;
*ppWriter = pDelFWriter;
return code;
_err:
tsdbError("vgId:%d, failed to open del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
*ppWriter = NULL;
return code;
}
int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync) {
int32_t code = 0;
SDelFWriter *pWriter = *ppWriter;
STsdb *pTsdb = pWriter->pTsdb;
// sync
if (sync && taosFsyncFile(pWriter->pWriteH) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// close
if (taosCloseFile(&pWriter->pWriteH) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree(pWriter->aBuf[iBuf]);
}
taosMemoryFree(pWriter);
*ppWriter = NULL;
return code;
_err:
tsdbError("vgId:%d, failed to close del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbWriteDelData(SDelFWriter *pWriter, SArray *aDelData, SDelIdx *pDelIdx) {
int32_t code = 0;
int64_t size;
int64_t n;
// prepare
size = sizeof(uint32_t);
for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
size += tPutDelData(NULL, taosArrayGet(aDelData, iDelData));
}
size += sizeof(TSCKSUM);
// alloc
code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err;
// build
n = 0;
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
n += tPutDelData(pWriter->aBuf[0] + n, taosArrayGet(aDelData, iDelData));
}
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size);
// write
n = taosWriteFile(pWriter->pWriteH, pWriter->aBuf[0], size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
ASSERT(n == size);
// update
pDelIdx->offset = pWriter->fDel.size;
pDelIdx->size = size;
pWriter->fDel.size += size;
return code;
_err:
tsdbError("vgId:%d, failed to write del data since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbWriteDelIdx(SDelFWriter *pWriter, SArray *aDelIdx) {
int32_t code = 0;
int64_t size;
int64_t n;
SDelIdx *pDelIdx;
// prepare
size = sizeof(uint32_t);
for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
size += tPutDelIdx(NULL, taosArrayGet(aDelIdx, iDelIdx));
}
size += sizeof(TSCKSUM);
// alloc
code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err;
// build
n = 0;
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
n += tPutDelIdx(pWriter->aBuf[0] + n, taosArrayGet(aDelIdx, iDelIdx));
}
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size);
// write
n = taosWriteFile(pWriter->pWriteH, pWriter->aBuf[0], size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// update
pWriter->fDel.offset = pWriter->fDel.size;
pWriter->fDel.size += size;
return code;
_err:
tsdbError("vgId:%d, write del idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbUpdateDelFileHdr(SDelFWriter *pWriter) {
int32_t code = 0; int32_t code = 0;
char hdr[TSDB_FHDR_SIZE];
int64_t size = TSDB_FHDR_SIZE;
int64_t n;
// build
memset(hdr, 0, size);
tPutDelFile(hdr, &pWriter->fDel);
taosCalcChecksumAppend(0, hdr, size);
// seek
if (taosLSeekFile(pWriter->pWriteH, 0, SEEK_SET) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// write pFD->pFD = taosOpenFile(path, opt);
n = taosWriteFile(pWriter->pWriteH, hdr, size); if (pFD->pFD == NULL) {
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _exit;
} }
return code; pFD->szPage = 4096;
pFD->pgno = 0;
_err: pFD->nBuf = 0;
tsdbError("vgId:%d, update del file hdr failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); pFD->pBuf = taosMemoryMalloc(pFD->szPage);
return code; if (pFD->pBuf == NULL) {
}
// SDelFReader ====================================================
struct SDelFReader {
STsdb *pTsdb;
SDelFile fDel;
TdFilePtr pReadH;
uint8_t *aBuf[1];
};
int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile, STsdb *pTsdb) {
int32_t code = 0;
char fname[TSDB_FILENAME_LEN];
SDelFReader *pDelFReader;
int64_t n;
// alloc
pDelFReader = (SDelFReader *)taosMemoryCalloc(1, sizeof(*pDelFReader));
if (pDelFReader == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _exit;
}
// open impl
pDelFReader->pTsdb = pTsdb;
pDelFReader->fDel = *pFile;
tsdbDelFileName(pTsdb, pFile, fname);
pDelFReader->pReadH = taosOpenFile(fname, TD_FILE_READ);
if (pDelFReader->pReadH == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
taosMemoryFree(pDelFReader);
goto _err;
} }
_exit: _exit:
*ppReader = pDelFReader;
return code; return code;
}
_err: void tsdbCloseFile(STsdbFD *pFD) {
tsdbError("vgId:%d, del file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); taosMemoryFree(pFD->pBuf);
*ppReader = NULL; taosCloseFile(&pFD->pFD);
return code;
} }
int32_t tsdbDelFReaderClose(SDelFReader **ppReader) { int32_t tsdbSyncFile(STsdbFD *pFD) {
int32_t code = 0; int32_t code = 0;
SDelFReader *pReader = *ppReader;
if (pReader) { if (taosFsyncFile(pFD->pFD) < 0) {
if (taosCloseFile(&pReader->pReadH) < 0) { code = TAOS_SYSTEM_ERROR(errno);
code = TAOS_SYSTEM_ERROR(errno); goto _exit;
goto _exit;
}
for (int32_t iBuf = 0; iBuf < sizeof(pReader->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree(pReader->aBuf[iBuf]);
}
taosMemoryFree(pReader);
} }
*ppReader = NULL;
_exit: _exit:
return code; return code;
} }
int32_t tsdbReadDelData(SDelFReader *pReader, SDelIdx *pDelIdx, SArray *aDelData) { int32_t tsdbWriteFile(STsdbFD *pFD, uint8_t *pBuf, int32_t nBuf, int64_t *offset) {
int32_t code = 0; int32_t code = 0;
int64_t offset = pDelIdx->offset;
int64_t size = pDelIdx->size;
int64_t n;
taosArrayClear(aDelData);
// seek
if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// alloc
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
// read int32_t n = 0;
n = taosReadFile(pReader->pReadH, pReader->aBuf[0], size); while (n < nBuf) {
if (n < 0) { int32_t remain = pFD->szPage - pFD->nBuf - sizeof(TSCKSUM);
code = TAOS_SYSTEM_ERROR(errno); int32_t size = TMIN(remain, nBuf - n);
goto _err;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
// check memcpy(pFD->pBuf + pFD->nBuf, pBuf + n, size);
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) { n += size;
code = TSDB_CODE_FILE_CORRUPTED; pFD->nBuf += size;
goto _err;
}
// // decode if (pFD->nBuf + sizeof(TSCKSUM) == pFD->szPage) {
n = 0; taosCalcChecksumAppend(0, pFD->pBuf, pFD->szPage);
uint32_t delimiter; int64_t n = taosWriteFile(pFD->pFD, pFD->pBuf, pFD->szPage);
n += tGetU32(pReader->aBuf[0] + n, &delimiter); if (n < 0) {
while (n < size - sizeof(TSCKSUM)) { code = TAOS_SYSTEM_ERROR(errno);
SDelData delData; goto _exit;
n += tGetDelData(pReader->aBuf[0] + n, &delData); }
if (taosArrayPush(aDelData, &delData) == NULL) { pFD->nBuf = 0;
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
} }
} }
ASSERT(n == size - sizeof(TSCKSUM)); _exit:
return code;
_err:
tsdbError("vgId:%d, read del data failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx) { static int32_t tsdbReadFilePage(STsdbFD *pFD, int64_t pgno) {
int32_t code = 0; int32_t code = 0;
int32_t n;
int64_t offset = pReader->fDel.offset;
int64_t size = pReader->fDel.size - offset;
taosArrayClear(aDelIdx);
// seek int64_t n = taosLSeekFile(pFD->pFD, pgno * pFD->szPage, SEEK_SET);
if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _exit;
} }
// alloc n = taosReadFile(pFD->pFD, pFD->pBuf, pFD->szPage);
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
// read
n = taosReadFile(pReader->pReadH, pReader->aBuf[0], size);
if (n < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _exit;
} else if (n < size) { } else if (n < pFD->szPage) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _exit;
} }
// check if (!taosCheckChecksumWhole(pFD->pBuf, pFD->szPage)) {
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _exit;
} }
// decode pFD->pgno = pgno;
n = 0;
uint32_t delimiter;
n += tGetU32(pReader->aBuf[0] + n, &delimiter);
ASSERT(delimiter == TSDB_FILE_DLMT);
while (n < size - sizeof(TSCKSUM)) { _exit:
SDelIdx delIdx; return code;
}
n += tGetDelIdx(pReader->aBuf[0] + n, &delIdx); int64_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t count) {
int32_t code = 0;
if (taosArrayPush(aDelIdx, &delIdx) == NULL) { int64_t pgno = offset / pFD->szPage;
code = TSDB_CODE_OUT_OF_MEMORY; int64_t n = 0;
goto _err; if (pFD->pgno == pgno) {
} int64_t bOff = offset % pFD->szPage;
int64_t nRead = TMIN(pFD->szPage - bOff - sizeof(TSCKSUM), count);
memcpy(pBuf + n, pFD->pBuf + bOff, nRead);
n = nRead;
} }
ASSERT(n == size - sizeof(TSCKSUM)); while (n < count) {
code = tsdbReadFilePage(pFD, pgno);
if (code) goto _exit;
return code; pgno++;
_err: int64_t nRead = TMIN(pFD->szPage - sizeof(TSCKSUM), count - n);
tsdbError("vgId:%d, read del idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code)); memcpy(pBuf + n, pFD->pBuf, nRead);
n += nRead;
}
_exit:
return code; return code;
} }
...@@ -1570,174 +1324,419 @@ int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) { ...@@ -1570,174 +1324,419 @@ int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) {
goto _err; goto _err;
} }
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->aSstF[0]->size); n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->aSstF[0]->size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// sma
tsdbSmaFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pSmaF, fNameFrom);
tsdbSmaFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pSmaF, fNameTo);
pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (pOutFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
if (PInFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pSmaF->size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
return code;
_err:
tsdbError("vgId:%d, tsdb DFileSet copy failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code;
}
// SDelFWriter ====================================================
int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb) {
int32_t code = 0;
char fname[TSDB_FILENAME_LEN];
char hdr[TSDB_FHDR_SIZE] = {0};
SDelFWriter *pDelFWriter;
int64_t n;
// alloc
pDelFWriter = (SDelFWriter *)taosMemoryCalloc(1, sizeof(*pDelFWriter));
if (pDelFWriter == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pDelFWriter->pTsdb = pTsdb;
pDelFWriter->fDel = *pFile;
tsdbDelFileName(pTsdb, pFile, fname);
pDelFWriter->pWriteH = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE);
if (pDelFWriter->pWriteH == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// update header
n = taosWriteFile(pDelFWriter->pWriteH, &hdr, TSDB_FHDR_SIZE);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
pDelFWriter->fDel.size = TSDB_FHDR_SIZE;
pDelFWriter->fDel.offset = 0;
*ppWriter = pDelFWriter;
return code;
_err:
tsdbError("vgId:%d, failed to open del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
*ppWriter = NULL;
return code;
}
int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync) {
int32_t code = 0;
SDelFWriter *pWriter = *ppWriter;
STsdb *pTsdb = pWriter->pTsdb;
// sync
if (sync && taosFsyncFile(pWriter->pWriteH) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// close
if (taosCloseFile(&pWriter->pWriteH) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree(pWriter->aBuf[iBuf]);
}
taosMemoryFree(pWriter);
*ppWriter = NULL;
return code;
_err:
tsdbError("vgId:%d, failed to close del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbWriteDelData(SDelFWriter *pWriter, SArray *aDelData, SDelIdx *pDelIdx) {
int32_t code = 0;
int64_t size;
int64_t n;
// prepare
size = sizeof(uint32_t);
for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
size += tPutDelData(NULL, taosArrayGet(aDelData, iDelData));
}
size += sizeof(TSCKSUM);
// alloc
code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err;
// build
n = 0;
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
n += tPutDelData(pWriter->aBuf[0] + n, taosArrayGet(aDelData, iDelData));
}
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size);
// write
n = taosWriteFile(pWriter->pWriteH, pWriter->aBuf[0], size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
ASSERT(n == size);
// update
pDelIdx->offset = pWriter->fDel.size;
pDelIdx->size = size;
pWriter->fDel.size += size;
return code;
_err:
tsdbError("vgId:%d, failed to write del data since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbWriteDelIdx(SDelFWriter *pWriter, SArray *aDelIdx) {
int32_t code = 0;
int64_t size;
int64_t n;
SDelIdx *pDelIdx;
// prepare
size = sizeof(uint32_t);
for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
size += tPutDelIdx(NULL, taosArrayGet(aDelIdx, iDelIdx));
}
size += sizeof(TSCKSUM);
// alloc
code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err;
// build
n = 0;
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
n += tPutDelIdx(pWriter->aBuf[0] + n, taosArrayGet(aDelIdx, iDelIdx));
}
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size);
// write
n = taosWriteFile(pWriter->pWriteH, pWriter->aBuf[0], size);
if (n < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// sma // update
tsdbSmaFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pSmaF, fNameFrom); pWriter->fDel.offset = pWriter->fDel.size;
tsdbSmaFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pSmaF, fNameTo); pWriter->fDel.size += size;
pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC); return code;
if (pOutFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ); _err:
if (PInFD == NULL) { tsdbError("vgId:%d, write del idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbUpdateDelFileHdr(SDelFWriter *pWriter) {
int32_t code = 0;
char hdr[TSDB_FHDR_SIZE];
int64_t size = TSDB_FHDR_SIZE;
int64_t n;
// build
memset(hdr, 0, size);
tPutDelFile(hdr, &pWriter->fDel);
taosCalcChecksumAppend(0, hdr, size);
// seek
if (taosLSeekFile(pWriter->pWriteH, 0, SEEK_SET) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pSmaF->size); // write
n = taosWriteFile(pWriter->pWriteH, hdr, size);
if (n < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
return code; return code;
_err: _err:
tsdbError("vgId:%d, tsdb DFileSet copy failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, update del file hdr failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
// =============== PAGE-WISE FILE =============== // SDelFReader ====================================================
typedef struct { struct SDelFReader {
TdFilePtr pFD; STsdb *pTsdb;
int32_t szPage; SDelFile fDel;
int32_t nBuf; TdFilePtr pReadH;
uint8_t *pBuf;
int64_t pgno;
} STsdbFD;
int32_t tsdbOpenFile(const char *path, int32_t opt, STsdbFD *pFD) { uint8_t *aBuf[1];
int32_t code = 0; };
pFD->pFD = taosOpenFile(path, opt); int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile, STsdb *pTsdb) {
if (pFD->pFD == NULL) { int32_t code = 0;
code = TAOS_SYSTEM_ERROR(errno); char fname[TSDB_FILENAME_LEN];
goto _exit; SDelFReader *pDelFReader;
} int64_t n;
pFD->szPage = 4096; // alloc
pFD->pgno = 0; pDelFReader = (SDelFReader *)taosMemoryCalloc(1, sizeof(*pDelFReader));
pFD->nBuf = 0; if (pDelFReader == NULL) {
pFD->pBuf = taosMemoryMalloc(pFD->szPage);
if (pFD->pBuf == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit; goto _err;
}
// open impl
pDelFReader->pTsdb = pTsdb;
pDelFReader->fDel = *pFile;
tsdbDelFileName(pTsdb, pFile, fname);
pDelFReader->pReadH = taosOpenFile(fname, TD_FILE_READ);
if (pDelFReader->pReadH == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
taosMemoryFree(pDelFReader);
goto _err;
} }
_exit: _exit:
*ppReader = pDelFReader;
return code; return code;
}
void tsdbCloseFile(STsdbFD *pFD) { _err:
taosMemoryFree(pFD->pBuf); tsdbError("vgId:%d, del file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
taosCloseFile(&pFD->pFD); *ppReader = NULL;
return code;
} }
int32_t tsdbSyncFile(STsdbFD *pFD) { int32_t tsdbDelFReaderClose(SDelFReader **ppReader) {
int32_t code = 0; int32_t code = 0;
SDelFReader *pReader = *ppReader;
if (taosFsyncFile(pFD->pFD) < 0) { if (pReader) {
code = TAOS_SYSTEM_ERROR(errno); if (taosCloseFile(&pReader->pReadH) < 0) {
goto _exit; code = TAOS_SYSTEM_ERROR(errno);
goto _exit;
}
for (int32_t iBuf = 0; iBuf < sizeof(pReader->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree(pReader->aBuf[iBuf]);
}
taosMemoryFree(pReader);
} }
*ppReader = NULL;
_exit: _exit:
return code; return code;
} }
int32_t tsdbWriteFile(STsdbFD *pFD, uint8_t *pBuf, int32_t nBuf, int64_t *offset) { int32_t tsdbReadDelData(SDelFReader *pReader, SDelIdx *pDelIdx, SArray *aDelData) {
int32_t code = 0; int32_t code = 0;
int64_t offset = pDelIdx->offset;
int64_t size = pDelIdx->size;
int64_t n;
int32_t n = 0; taosArrayClear(aDelData);
while (n < nBuf) {
int32_t remain = pFD->szPage - pFD->nBuf - sizeof(TSCKSUM);
int32_t size = TMIN(remain, nBuf - n);
memcpy(pFD->pBuf + pFD->nBuf, pBuf + n, size); // seek
n += size; if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
pFD->nBuf += size; code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (pFD->nBuf + sizeof(TSCKSUM) == pFD->szPage) { // alloc
taosCalcChecksumAppend(0, pFD->pBuf, pFD->szPage); code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
int64_t n = taosWriteFile(pFD->pFD, pFD->pBuf, pFD->szPage); // read
if (n < 0) { n = taosReadFile(pReader->pReadH, pReader->aBuf[0], size);
code = TAOS_SYSTEM_ERROR(errno); if (n < 0) {
goto _exit; code = TAOS_SYSTEM_ERROR(errno);
} goto _err;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
pFD->nBuf = 0; // check
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
// // decode
n = 0;
uint32_t delimiter;
n += tGetU32(pReader->aBuf[0] + n, &delimiter);
while (n < size - sizeof(TSCKSUM)) {
SDelData delData;
n += tGetDelData(pReader->aBuf[0] + n, &delData);
if (taosArrayPush(aDelData, &delData) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
} }
} }
_exit: ASSERT(n == size - sizeof(TSCKSUM));
return code;
_err:
tsdbError("vgId:%d, read del data failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
static int32_t tsdbReadFilePage(STsdbFD *pFD, int64_t pgno) { int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx) {
int32_t code = 0; int32_t code = 0;
int32_t n;
int64_t offset = pReader->fDel.offset;
int64_t size = pReader->fDel.size - offset;
int64_t n = taosLSeekFile(pFD->pFD, pgno * pFD->szPage, SEEK_SET); taosArrayClear(aDelIdx);
if (n < 0) {
// seek
if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _exit; goto _err;
} }
n = taosReadFile(pFD->pFD, pFD->pBuf, pFD->szPage); // alloc
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
// read
n = taosReadFile(pReader->pReadH, pReader->aBuf[0], size);
if (n < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _exit; goto _err;
} else if (n < pFD->szPage) { } else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _exit; goto _err;
} }
if (!taosCheckChecksumWhole(pFD->pBuf, pFD->szPage)) { // check
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _exit; goto _err;
} }
pFD->pgno = pgno; // decode
n = 0;
uint32_t delimiter;
n += tGetU32(pReader->aBuf[0] + n, &delimiter);
ASSERT(delimiter == TSDB_FILE_DLMT);
_exit: while (n < size - sizeof(TSCKSUM)) {
return code; SDelIdx delIdx;
}
int64_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t count) { n += tGetDelIdx(pReader->aBuf[0] + n, &delIdx);
int32_t code = 0;
int64_t pgno = offset / pFD->szPage; if (taosArrayPush(aDelIdx, &delIdx) == NULL) {
int64_t n = 0; code = TSDB_CODE_OUT_OF_MEMORY;
if (pFD->pgno == pgno) { goto _err;
int64_t bOff = offset % pFD->szPage; }
int64_t nRead = TMIN(pFD->szPage - bOff - sizeof(TSCKSUM), count);
memcpy(pBuf + n, pFD->pBuf + bOff, nRead);
n = nRead;
} }
while (n < count) { ASSERT(n == size - sizeof(TSCKSUM));
code = tsdbReadFilePage(pFD, pgno);
if (code) goto _exit;
pgno++;
int64_t nRead = TMIN(pFD->szPage - sizeof(TSCKSUM), count - n); return code;
memcpy(pBuf + n, pFD->pBuf, nRead);
n += nRead;
}
_exit: _err:
tsdbError("vgId:%d, read del idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
\ No newline at end of file
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或