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

Merge pull request #12400 from taosdata/feature/tq 

fix(wal): int overflow
This commit is contained in:
Liu Jicong 2022-05-12 21:33:25 +08:00 committed by GitHub
parent c4c81c83af af2d2db5d3
commit 2cc8ff312d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 17 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -485,8 +485,10 @@ static int32_t mndProcessDropTopicReq(SNodeMsg *pReq) {
return -1;
}
}
// TODO: check ref
int32_t code = mndDropTopic(pMnode, pReq, pTopic);
// TODO: iterate and drop related subscriptions and offsets
mndReleaseTopic(pMnode, pTopic);
if (code != 0) {

8
source/libs/wal/src/walRead.c
View File

@ -55,7 +55,7 @@ int32_t walRegisterRead(SWalReadHandle *pRead, int64_t ver) {
}
static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, int64_t ver) {
int ret = 0;
int64_t ret = 0;
TdFilePtr pIdxTFile = pRead->pReadIdxTFile;
TdFilePtr pLogTFile = pRead->pReadLogTFile;
@ -68,14 +68,14 @@ static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, i
wError("failed to seek idx file, ver %ld, pos: %ld, since %s", ver, offset, terrstr());
return -1;
}
SWalIdxEntry entry;
SWalIdxEntry entry = {0};
if ((ret = taosReadFile(pIdxTFile, &entry, sizeof(SWalIdxEntry))) != sizeof(SWalIdxEntry)) {
if (ret < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
wError("failed to read idx file, since %s", terrstr());
} else {
terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
wError("read idx file incompletely, read bytes %d, bytes should be %lu", ret, sizeof(SWalIdxEntry));
wError("read idx file incompletely, read bytes %ld, bytes should be %lu", ret, sizeof(SWalIdxEntry));
}
return -1;
}
@ -187,7 +187,7 @@ int32_t walFetchHead(SWalReadHandle *pRead, int64_t ver, SWalHead *pHead) {
}
int32_t walSkipFetchBody(SWalReadHandle *pRead, const SWalHead *pHead) {
int32_t code;
int64_t code;
ASSERT(pRead->curVersion == pHead->head.version);

4
source/libs/wal/src/walWrite.c
View File

@ -30,8 +30,8 @@ int32_t walCommit(SWal *pWal, int64_t ver) {
}
int32_t walRollback(SWal *pWal, int64_t ver) {
int code;
char fnameStr[WAL_FILE_LEN];
int64_t code;
char fnameStr[WAL_FILE_LEN];
if (ver > pWal->vers.lastVer || ver < pWal->vers.commitVer) {
terrno = TSDB_CODE_WAL_INVALID_VER;
return -1;

36
source/os/src/osFile.c
View File

@ -246,11 +246,11 @@ TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
access |= (tdFileOptions & TD_FILE_APPEND) ? O_APPEND : 0;
access |= (tdFileOptions & TD_FILE_TEXT) ? O_TEXT : 0;
access |= (tdFileOptions & TD_FILE_EXCL) ? O_EXCL : 0;
#ifdef WINDOWS
fd = _open(path, access, _S_IREAD|_S_IWRITE);
#else
#ifdef WINDOWS
fd = _open(path, access, _S_IREAD | _S_IWRITE);
#else
fd = open(path, access, S_IRWXU | S_IRWXG | S_IRWXO);
#endif
#endif
if (fd == -1) {
return NULL;
}
@ -310,9 +310,6 @@ int64_t taosCloseFile(TdFilePtr *ppFile) {
}
int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) {
if (pFile == NULL) {
return 0;
}
#if FILE_WITH_LOCK
taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
@ -356,10 +353,10 @@ int64_t taosPReadFile(TdFilePtr pFile, void *buf, int64_t count, int64_t offset)
#if FILE_WITH_LOCK
taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
assert(pFile->fd >= 0); // Please check if you have closed the file.
assert(pFile->fd >= 0); // Please check if you have closed the file.
#ifdef WINDOWS
size_t pos = lseek(pFile->fd, 0, SEEK_CUR);
lseek(pFile->fd, (long)offset, SEEK_SET);
lseek(pFile->fd, offset, SEEK_SET);
int64_t ret = read(pFile->fd, buf, count);
lseek(pFile->fd, pos, SEEK_SET);
#else
@ -372,9 +369,6 @@ int64_t taosPReadFile(TdFilePtr pFile, void *buf, int64_t count, int64_t offset)
}
int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) {
if (pFile == NULL) {
return 0;
}
#if FILE_WITH_LOCK
taosThreadRwlockWrlock(&(pFile->rwlock));
#endif
@ -406,14 +400,11 @@ int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) {
}
int64_t taosLSeekFile(TdFilePtr pFile, int64_t offset, int32_t whence) {
if (pFile == NULL) {
return 0;
}
#if FILE_WITH_LOCK
taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
assert(pFile->fd >= 0); // Please check if you have closed the file.
int64_t ret = lseek(pFile->fd, (long)offset, whence);
int64_t ret = lseek(pFile->fd, offset, whence);
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
@ -424,9 +415,6 @@ int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime) {
#ifdef WINDOWS
return 0;
#else
if (pFile == NULL) {
return 0;
}
assert(pFile->fd >= 0); // Please check if you have closed the file.
struct stat fileStat;
@ -451,9 +439,6 @@ int32_t taosLockFile(TdFilePtr pFile) {
#ifdef WINDOWS
return 0;
#else
if (pFile == NULL) {
return 0;
}
assert(pFile->fd >= 0); // Please check if you have closed the file.
return (int32_t)flock(pFile->fd, LOCK_EX | LOCK_NB);
@ -464,9 +449,6 @@ int32_t taosUnLockFile(TdFilePtr pFile) {
#ifdef WINDOWS
return 0;
#else
if (pFile == NULL) {
return 0;
}
assert(pFile->fd >= 0); // Please check if you have closed the file.
return (int32_t)flock(pFile->fd, LOCK_UN | LOCK_NB);
@ -667,7 +649,7 @@ int32_t taosUmaskFile(int32_t maskVal) {
int32_t taosGetErrorFile(TdFilePtr pFile) { return errno; }
int64_t taosGetLineFile(TdFilePtr pFile, char **__restrict ptrBuf) {
if (pFile == NULL || ptrBuf == NULL ) {
if (pFile == NULL || ptrBuf == NULL) {
return -1;
}
if (*ptrBuf != NULL) {
@ -689,7 +671,7 @@ int64_t taosGetLineFile(TdFilePtr pFile, char **__restrict ptrBuf) {
#endif
}
int64_t taosGetsFile(TdFilePtr pFile, int32_t maxSize, char *__restrict buf) {
if (pFile == NULL || buf == NULL ) {
if (pFile == NULL || buf == NULL) {
return -1;
}
assert(pFile->fp != NULL);