Skip to content

TDengine
TDengine

yiicenne / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
TDengine
TDengine

体验新版 GitCode，发现更多精彩内容 >>

提交 a64fb809 编写于 作者: H Haojun Liao
浏览文件
操作

Merge branch 'develop' into feature/query

上级 4f5c5887 f2d06f08
隐藏空白更改
内联 并排
Showing with 460 addition and 316 deletion
.travis.yml
浏览文件 @ a64fb809
......@@ -135,92 +135,6 @@ matrix:
# https://scan.coverity.com/faq#frequency
branch_pattern: coverity_scan
- os: linux
dist: bionic
language: c
compiler: gcc
env: ENV_COVER=true
git:
- depth: 1
addons:
apt:
packages:
- build-essential
- cmake
- net-tools
- python-pip
- python-setuptools
- python3-pip
- python3-setuptools
- lcov
- psmisc
before_script:
- cd ${TRAVIS_BUILD_DIR}
- mkdir debug
- cd debug
script:
- cmake -DCOVER=true .. > /dev/null
- make > /dev/null
after_success:
- |-
case $TRAVIS_OS_NAME in
linux)
cd ${TRAVIS_BUILD_DIR}/debug
make install > /dev/null || travis_terminate $?
pip install numpy
pip install --user ${TRAVIS_BUILD_DIR}/src/connector/python/linux/python2/
pip3 install numpy
pip3 install --user ${TRAVIS_BUILD_DIR}/src/connector/python/linux/python3/
cd ${TRAVIS_BUILD_DIR}/tests
./test-all.sh smoke COVER
TEST_RESULT=$?
pkill taosd
sleep 1
cd ${TRAVIS_BUILD_DIR}
lcov -d . --capture --rc lcov_branch_coverage=1 -o coverage.info
lcov --remove coverage.info '*/tests/*' '*/test/*' '*/deps/*' '*/plugins/*' -o coverage.info
lcov -l --rc lcov_branch_coverage=1 coverage.info || travis_terminate $?
gem install coveralls-lcov
# Color setting
RED='\033[0;31m'
GREEN='\033[1;32m'
GREEN_DARK='\033[0;32m'
GREEN_UNDERLINE='\033[4;32m'
NC='\033[0m'
coveralls-lcov coverage.info
if [ "$?" -eq "0" ]; then
echo -e "${GREEN} ## Uploaded to Coveralls.io! ## ${NC}"
else
echo -e "${RED} ## Coveralls.io not collect coverage report! ## ${NC} "
fi
bash <(curl -s https://codecov.io/bash) -y .codecov.yml -f coverage.info
if [ "$?" -eq "0" ]; then
echo -e "${GREEN} ## Uploaded to Codecov! ## ${NC} "
else
echo -e "${RED} ## Codecov did not collect coverage report! ## ${NC} "
fi
if [ "$TEST_RESULT" -ne "0" ]; then
travis_terminate $?
fi
;;
esac
- os: linux
dist: trusty
language: c
......
README.md
浏览文件 @ a64fb809
[![Build Status](https://travis-ci.org/taosdata/TDengine.svg?branch=master)](https://travis-ci.org/taosdata/TDengine)
[![Build status](https://ci.appveyor.com/api/projects/status/kf3pwh2or5afsgl9/branch/master?svg=true)](https://ci.appveyor.com/project/sangshuduo/tdengine-2n8ge/branch/master)
[![Coverage Status](https://coveralls.io/repos/github/taosdata/TDengine/badge.svg?branch=develop)](https://coveralls.io/github/taosdata/TDengine?branch=develop)
[![TDengine](TDenginelogo.png)](https://www.taosdata.com)
......
src/client/src/tscParseInsert.c
浏览文件 @ a64fb809
......@@ -46,18 +46,20 @@ static int32_t tscToInteger(SSQLToken *pToken, int64_t *value, char **endPtr) {
return TK_ILLEGAL;
}
int32_t radix = 10;
int32_t radixList[3] = {16, 8, 2}; // the integer number with different radix: hex, oct, bin
if (pToken->type == TK_HEX || pToken->type == TK_OCT || pToken->type == TK_BIN) {
radix = radixList[pToken->type - TK_HEX];
}
errno = 0;
*value = strtoll(pToken->z, endPtr, radix);
*value = strtoll(pToken->z, endPtr, 0);
if (**endPtr == 'e' || **endPtr == 'E' || **endPtr == '.') {
errno = 0;
double v = round(strtod(pToken->z, endPtr));
if (v > INT64_MAX || v <= INT64_MIN) {
errno = ERANGE;
} else {
*value = v;
}
}
// not a valid integer number, return error
if ((pToken->type == TK_STRING || pToken->type == TK_ID) && ((*endPtr - pToken->z) != pToken->n)) {
if (*endPtr - pToken->z != pToken->n) {
return TK_ILLEGAL;
}
......@@ -73,11 +75,11 @@ static int32_t tscToDouble(SSQLToken *pToken, double *value, char **endPtr) {
*value = strtod(pToken->z, endPtr);
// not a valid integer number, return error
if ((pToken->type == TK_STRING || pToken->type == TK_ID) && ((*endPtr - pToken->z) != pToken->n)) {
if ((*endPtr - pToken->z) != pToken->n) {
return TK_ILLEGAL;
} else {
return pToken->type;
}
return pToken->type;
}
int tsParseTime(SSQLToken *pToken, int64_t *time, char **next, char *error, int16_t timePrec) {
......@@ -986,14 +988,14 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
return code;
}
int validateTableName(char *tblName, int len) {
char buf[TSDB_TABLE_ID_LEN] = {0};
tstrncpy(buf, tblName, sizeof(buf));
int validateTableName(char *tblName, int len, SSQLToken* psTblToken) {
tstrncpy(psTblToken->z, tblName, TSDB_TABLE_ID_LEN);
SSQLToken token = {.n = len, .type = TK_ID, .z = buf};
tSQLGetToken(buf, &token.type);
psTblToken->n = len;
psTblToken->type = TK_ID;
tSQLGetToken(psTblToken->z, &psTblToken->type);
return tscValidateName(&token);
return tscValidateName(psTblToken);
}
static int32_t validateDataSource(SSqlCmd *pCmd, int8_t type, const char *sql) {
......@@ -1076,14 +1078,16 @@ int tsParseInsertSql(SSqlObj *pSql) {
}
pCmd->curSql = sToken.z;
char buf[TSDB_TABLE_ID_LEN];
SSQLToken sTblToken;
sTblToken.z = buf;
// Check if the table name available or not
if (validateTableName(sToken.z, sToken.n) != TSDB_CODE_SUCCESS) {
if (validateTableName(sToken.z, sToken.n, &sTblToken) != TSDB_CODE_SUCCESS) {
code = tscInvalidSQLErrMsg(pCmd->payload, "table name invalid", sToken.z);
goto _error;
}
if ((code = tscSetTableFullName(pTableMetaInfo, &sToken, pSql)) != TSDB_CODE_SUCCESS) {
if ((code = tscSetTableFullName(pTableMetaInfo, &sTblToken, pSql)) != TSDB_CODE_SUCCESS) {
goto _error;
}
......
src/inc/ttokendef.h
浏览文件 @ a64fb809
......@@ -221,6 +221,16 @@
#define TK_INTO 203
#define TK_VALUES 204
#define TK_SPACE 300
#define TK_COMMENT 301
#define TK_ILLEGAL 302
#define TK_HEX 303 // hex number 0x123
#define TK_OCT 304 // oct number
#define TK_BIN 305 // bin format data 0b111
#define TK_FILE 306
#define TK_QUESTION 307 // denoting the placeholder of "?",when invoking statement bind query
#endif
src/kit/shell/src/shellEngine.c
浏览文件 @ a64fb809
......@@ -474,7 +474,7 @@ static int dumpResultToFile(const char* fname, TAOS_RES* tres) {
} while( row != NULL);
result = NULL;
taos_free_result(tres);
//taos_free_result(tres);
fclose(fp);
return numOfRows;
......@@ -803,6 +803,7 @@ void source_file(TAOS *con, char *fptr) {
char *fname = full_path.we_wordv[0];
/*
if (access(fname, F_OK) != 0) {
fprintf(stderr, "ERROR: file %s is not exist\n", fptr);
......@@ -810,6 +811,7 @@ void source_file(TAOS *con, char *fptr) {
free(cmd);
return;
}
*/
FILE *f = fopen(fname, "r");
if (f == NULL) {
......@@ -849,7 +851,7 @@ void source_file(TAOS *con, char *fptr) {
void shellGetGrantInfo(void *con) {
return;
#if 0
char sql[] = "show grants";
TAOS_RES* tres = taos_query(con, sql);
......@@ -900,4 +902,5 @@ void shellGetGrantInfo(void *con) {
}
fprintf(stdout, "\n");
#endif
}
src/kit/shell/src/shellImport.c
浏览文件 @ a64fb809
......@@ -73,7 +73,7 @@ static void shellParseDirectory(const char *directoryName, const char *prefix, c
}
int fileNum = 0;
while (fscanf(fp, "%s", fileArray[fileNum++])) {
while (fscanf(fp, "%128s", fileArray[fileNum++])) {
if (strcmp(fileArray[fileNum-1], shellTablesSQLFile) == 0) {
fileNum--;
}
......@@ -150,9 +150,11 @@ static void shellSourceFile(TAOS *con, char *fptr) {
char *fname = full_path.we_wordv[0];
if (fname == NULL) {
fprintf(stderr, "ERROR: invalid filename\n");
free(cmd);
return;
}
/*
if (access(fname, F_OK) != 0) {
fprintf(stderr, "ERROR: file %s is not exist\n", fptr);
......@@ -168,6 +170,7 @@ static void shellSourceFile(TAOS *con, char *fptr) {
free(cmd);
return;
}
*/
FILE *f = fopen(fname, "r");
if (f == NULL) {
......
src/kit/shell/src/shellLinux.c
浏览文件 @ a64fb809
......@@ -162,7 +162,7 @@ void shellReadCommand(TAOS *con, char *command) {
// Read input.
char c;
while (1) {
c = getchar();
c = (char)getchar(); // getchar() return an 'int' value
if (c < 0) { // For UTF-8
int count = countPrefixOnes(c);
......
src/kit/taosdemo/taosdemo.c
浏览文件 @ a64fb809
......@@ -34,6 +34,7 @@
#include <wordexp.h>
#include "taos.h"
#include "tutil.h"
extern char configDir[];
......@@ -82,7 +83,7 @@ typedef struct DemoArguments {
bool insert_only;
char *output_file;
int mode;
char *datatype[MAX_NUM_DATATYPE];
char *datatype[MAX_NUM_DATATYPE+1];
int len_of_binary;
int num_of_CPR;
int num_of_threads;
......@@ -432,7 +433,7 @@ int main(int argc, char *argv[]) {
tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec);
printf("###################################################################\n\n");
printf("Press enter key to continue");
getchar();
(void)getchar();
fprintf(fp, "###################################################################\n");
fprintf(fp, "# Server IP: %s:%hu\n", ip_addr == NULL ? "localhost" : ip_addr, port);
......@@ -845,10 +846,10 @@ void *syncWrite(void *sarg) {
pstr += sprintf(pstr, "insert into %s.%s%d values", winfo->db_name, winfo->tb_prefix, tID);
int k;
for (k = 0; k < winfo->nrecords_per_request;) {
int rand_num = rand() % 100;
int rand_num = trand() % 100;
int len = -1;
if (winfo->data_of_order ==1 && rand_num < winfo->data_of_rate) {
long d = tmp_time - rand() % 1000000 + rand_num;
long d = tmp_time - trand() % 1000000 + rand_num;
len = generateData(data, data_type, ncols_per_record, d, len_of_binary);
} else {
len = generateData(data, data_type, ncols_per_record, tmp_time += 1000, len_of_binary);
......@@ -940,10 +941,10 @@ void callBack(void *param, TAOS_RES *res, int code) {
pstr += sprintf(pstr, "insert into %s values", tb_info->tb_name);
for (int i = 0; i < tb_info->nrecords_per_request; i++) {
int rand_num = rand() % 100;
int rand_num = trand() % 100;
if (tb_info->data_of_order ==1 && rand_num < tb_info->data_of_rate)
{
long d = tmp_time - rand() % 1000000 + rand_num;
long d = tmp_time - trand() % 1000000 + rand_num;
generateData(data, datatype, ncols_per_record, d, len_of_binary);
} else
{
......@@ -985,22 +986,27 @@ int32_t generateData(char *res, char **data_type, int num_of_cols, int64_t times
}
}
if (0 == c) {
perror("data type error!");
exit(-1);
}
for (int i = 0; i < num_of_cols; i++) {
if (strcasecmp(data_type[i % c], "tinyint") == 0) {
pstr += sprintf(pstr, ", %d", (int)(rand() % 128));
pstr += sprintf(pstr, ", %d", (int)(trand() % 128));
} else if (strcasecmp(data_type[i % c], "smallint") == 0) {
pstr += sprintf(pstr, ", %d", (int)(rand() % 32767));
pstr += sprintf(pstr, ", %d", (int)(trand() % 32767));
} else if (strcasecmp(data_type[i % c], "int") == 0) {
pstr += sprintf(pstr, ", %d", (int)(rand() % 10));
pstr += sprintf(pstr, ", %d", (int)(trand() % 10));
} else if (strcasecmp(data_type[i % c], "bigint") == 0) {
pstr += sprintf(pstr, ", %" PRId64, rand() % 2147483648);
pstr += sprintf(pstr, ", %" PRId64, trand() % 2147483648);
} else if (strcasecmp(data_type[i % c], "float") == 0) {
pstr += sprintf(pstr, ", %10.4f", (float)(rand() / 1000));
pstr += sprintf(pstr, ", %10.4f", (float)(trand() / 1000));
} else if (strcasecmp(data_type[i % c], "double") == 0) {
double t = (double)(rand() / 1000000);
double t = (double)(trand() / 1000000);
pstr += sprintf(pstr, ", %20.8f", t);
} else if (strcasecmp(data_type[i % c], "bool") == 0) {
bool b = rand() & 1;
bool b = trand() & 1;
pstr += sprintf(pstr, ", %s", b ? "true" : "false");
} else if (strcasecmp(data_type[i % c], "binary") == 0) {
char s[len_of_binary];
......@@ -1026,7 +1032,7 @@ void rand_string(char *str, int size) {
--size;
int n;
for (n = 0; n < size; n++) {
int key = rand() % (int)(sizeof charset - 1);
int key = trand() % (int)(sizeof charset - 1);
str[n] = charset[key];
}
str[n] = 0;
......
src/kit/taosdump/taosdump.c
浏览文件 @ a64fb809
......@@ -229,7 +229,7 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
fprintf(stderr, "Invalid path %s\n", arg);
return -1;
}
strcpy(arguments->output, full_path.we_wordv[0]);
tstrncpy(arguments->output, full_path.we_wordv[0], TSDB_FILENAME_LEN);
wordfree(&full_path);
break;
case 'i':
......@@ -411,7 +411,7 @@ int taosGetTableRecordInfo(char *table, STableRecordInfo *pTableRecordInfo) {
if ((row = taos_fetch_row(result)) != NULL) {
isSet = true;
pTableRecordInfo->isMetric = true;
strcpy(pTableRecordInfo->tableRecord.metric, table);
tstrncpy(pTableRecordInfo->tableRecord.metric, table, TSDB_TABLE_NAME_LEN);
}
taos_free_result(result);
......@@ -642,9 +642,12 @@ int taosDumpDb(SDbInfo *dbInfo, SDumpArguments *arguments, FILE *fp) {
taos_free_result(result);
lseek(fd, 0, SEEK_SET);
(void)lseek(fd, 0, SEEK_SET);
while (read(fd, &tableRecord, sizeof(STableRecord)) > 0) {
while (1) {
ssize_t ret = read(fd, &tableRecord, sizeof(STableRecord));
if (ret <= 0) break;
tableRecord.name[sizeof(tableRecord.name) - 1] = 0;
tableRecord.metric[sizeof(tableRecord.metric) - 1] = 0;
taosDumpTable(tableRecord.name, tableRecord.metric, arguments, fp);
......@@ -807,7 +810,7 @@ int taosGetTableDes(char *table, STableDef *tableDes) {
TAOS_FIELD *fields = taos_fetch_fields(result);
strcpy(tableDes->name, table);
tstrncpy(tableDes->name, table, TSDB_COL_NAME_LEN);
while ((row = taos_fetch_row(result)) != NULL) {
strncpy(tableDes->cols[count].field, (char *)row[TSDB_DESCRIBE_METRIC_FIELD_INDEX],
......@@ -903,16 +906,19 @@ int32_t taosDumpMetric(char *metric, SDumpArguments *arguments, FILE *fp) {
taos_free_result(result);
result = NULL;
lseek(fd, 0, SEEK_SET);
(void)lseek(fd, 0, SEEK_SET);
while (read(fd, &tableRecord, sizeof(STableRecord)) > 0) {
while (1) {
ssize_t ret = read(fd, &tableRecord, sizeof(STableRecord));
if (ret <= 0) break;
tableRecord.name[sizeof(tableRecord.name) - 1] = 0;
tableRecord.metric[sizeof(tableRecord.metric) - 1] = 0;
taosDumpTable(tableRecord.name, tableRecord.metric, arguments, fp);
}
tclose(fd);
remove(".table.tmp");
(void)remove(".table.tmp");
return 0;
}
......@@ -1004,7 +1010,7 @@ int taosDumpTableData(FILE *fp, char *tbname, SDumpArguments *arguments) {
break;
}
}
pstr += sprintf(pstr, ")");
sprintf(pstr, ")");
count++;
fprintf(fp, "%s", buffer);
......@@ -1327,7 +1333,7 @@ int convertNCharToReadable(char *str, int size, char *buf, int bufsize) {
if ((int)wc < 256) {
pbuf = stpcpy(pbuf, ascii_literal_list[(int)wc]);
} else {
} else if (byte_width > 0) {
memcpy(pbuf, pstr, byte_width);
pbuf += byte_width;
}
......
src/mnode/src/mnodeAcct.c
浏览文件 @ a64fb809
......@@ -126,8 +126,8 @@ int32_t mnodeInitAccts() {
}
void mnodeCleanupAccts() {
sdbCloseTable(tsAcctSdb);
acctCleanUp();
sdbCloseTable(tsAcctSdb);
}
void *mnodeGetAcct(char *name) {
......
src/mnode/src/mnodeDnode.c
浏览文件 @ a64fb809
......@@ -348,7 +348,6 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
pRsp->dnodeCfg.dnodeId = htonl(pDnode->dnodeId);
pRsp->dnodeCfg.moduleStatus = htonl((int32_t)pDnode->isMgmt);
pRsp->dnodeCfg.numOfVnodes = htonl(openVnodes);
mnodeGetMnodeInfos(&pRsp->mnodes);
SDMVgroupAccess *pAccess = (SDMVgroupAccess *)((char *)pRsp + sizeof(SDMStatusRsp));
for (int32_t j = 0; j < openVnodes; ++j) {
......@@ -392,6 +391,10 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
}
pDnode->lastAccess = tsAccessSquence;
//this func should be called after sdb replica changed
mnodeGetMnodeInfos(&pRsp->mnodes);
mnodeDecDnodeRef(pDnode);
pMsg->rpcRsp.len = contLen;
......
src/mnode/src/mnodeVgroup.c
浏览文件 @ a64fb809
......@@ -678,8 +678,9 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
SMnodeMsg *mnodeMsg = rpcMsg->handle;
mnodeMsg->received++;
if (rpcMsg->code == TSDB_CODE_SUCCESS) {
mnodeMsg->code = rpcMsg->code;
mnodeMsg->successed++;
} else {
mnodeMsg->code = rpcMsg->code;
}
SVgObj *pVgroup = mnodeMsg->pVgroup;
......@@ -702,7 +703,7 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
code = TSDB_CODE_MND_SDB_ERROR;
}
dnodeSendRpcMnodeWriteRsp(mnodeMsg, code);
dnodeSendRpcMnodeWriteRsp(mnodeMsg, mnodeMsg->code);
}
}
......
src/os/linux/src/linuxPlatform.c
浏览文件 @ a64fb809
......@@ -56,70 +56,8 @@ void taosMsleep(int mseconds) {
bool taosCheckPthreadValid(pthread_t thread) { return thread != 0; }
void taosResetPthread(pthread_t *thread) { *thread = 0; }
int64_t taosGetPthreadId() { return (int64_t)pthread_self(); }
/*
* Function to get the private ip address of current machine. If get IP
* successfully, return 0, else, return -1. The return values is ip.
*
* Use:
* if (taosGetPrivateIp(ip) != 0) {
* perror("Fail to get private IP address\n");
* exit(EXIT_FAILURE);
* }
*/
int taosGetPrivateIp(char *const ip) {
bool hasLoCard = false;
struct ifaddrs *ifaddr, *ifa;
int family, s;
char host[NI_MAXHOST];
if (getifaddrs(&ifaddr) == -1) {
return -1;
}
/* Walk through linked list, maintaining head pointer so we can free list later */
int flag = 0;
for (ifa = ifaddr; ifa != NULL; ifa = ifa->ifa_next) {
if (ifa->ifa_addr == NULL) continue;
family = ifa->ifa_addr->sa_family;
if (strcmp("lo", ifa->ifa_name) == 0) {
hasLoCard = true;
continue;
}
if (family == AF_INET) {
/* printf("%-8s", ifa->ifa_name); */
s = getnameinfo(ifa->ifa_addr, (family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6),
host, NI_MAXHOST, NULL, 0, NI_NUMERICHOST);
if (s != 0) {
freeifaddrs(ifaddr);
return -1;
}
strcpy(ip, host);
flag = 1;
break;
}
}
freeifaddrs(ifaddr);
if (flag) {
return 0;
} else {
if (hasLoCard) {
uPrint("no net card was found, use lo:127.0.0.1 as default");
strcpy(ip, "127.0.0.1");
return 0;
}
return -1;
}
}
int taosSetNonblocking(int sock, int on) {
int flags = 0;
if ((flags = fcntl(sock, F_GETFL, 0)) < 0) {
......@@ -294,21 +232,6 @@ ssize_t twrite(int fd, void *buf, size_t n) {
return n;
}
bool taosSkipSocketCheck() {
struct utsname buf;
if (uname(&buf)) {
uPrint("can't fetch os info");
return false;
}
if (strstr(buf.release, "Microsoft") != 0) {
uPrint("using WSLv1");
return true;
}
return false;
}
void taosBlockSIGPIPE() {
sigset_t signal_mask;
sigemptyset(&signal_mask);
......
src/os/linux/src/linuxSysPara.c
浏览文件 @ a64fb809
......@@ -162,7 +162,7 @@ static void taosGetSystemTimezone() {
FILE *f = fopen("/etc/timezone", "r");
char buf[65] = {0};
if (f != NULL) {
fread(buf, 64, 1, f);
(void)fread(buf, 64, 1, f);
fclose(f);
}
......@@ -547,7 +547,7 @@ void taosSetCoreDump() {
struct rlimit rlim;
struct rlimit rlim_new;
if (getrlimit(RLIMIT_CORE, &rlim) == 0) {
uPrint("the old unlimited para: rlim_cur=%d, rlim_max=%d", rlim.rlim_cur, rlim.rlim_max);
uPrint("the old unlimited para: rlim_cur=%d, rlim_max=%" PRIu64, rlim.rlim_cur, rlim.rlim_max);
rlim_new.rlim_cur = RLIM_INFINITY;
rlim_new.rlim_max = RLIM_INFINITY;
if (setrlimit(RLIMIT_CORE, &rlim_new) != 0) {
......@@ -559,7 +559,7 @@ void taosSetCoreDump() {
}
if (getrlimit(RLIMIT_CORE, &rlim) == 0) {
uPrint("the new unlimited para: rlim_cur=%d, rlim_max=%d", rlim.rlim_cur, rlim.rlim_max);
uPrint("the new unlimited para: rlim_cur=%d, rlim_max=%" PRIu64, rlim.rlim_cur, rlim.rlim_max);
}
#ifndef _TD_ARM_
......@@ -586,7 +586,7 @@ void taosSetCoreDump() {
uPrint("_sysctl(kern_core_uses_pid) set fail: %s", strerror(errno));
}
uPrint("The old core_uses_pid[%d]: %d", old_len, old_usespid);
uPrint("The old core_uses_pid[%" PRIu64 "]: %d", old_len, old_usespid);
old_usespid = 0;
......@@ -603,7 +603,7 @@ void taosSetCoreDump() {
uPrint("_sysctl(kern_core_uses_pid) get fail: %s", strerror(errno));
}
uPrint("The new core_uses_pid[%d]: %d", old_len, old_usespid);
uPrint("The new core_uses_pid[%" PRIu64 "]: %d", old_len, old_usespid);
#endif
#if 0
......
src/query/src/qtokenizer.c
浏览文件 @ a64fb809
......@@ -25,7 +25,7 @@
// All the keywords of the SQL language are stored in a hash table
typedef struct SKeyword {
const char* name; // The keyword name
uint8_t type; // type
uint16_t type; // type
uint8_t len; // length
} SKeyword;
......
src/rpc/inc/rpcTcp.h
浏览文件 @ a64fb809
......@@ -21,9 +21,11 @@ extern "C" {
#endif
void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThreads, void *fp, void *shandle);
void taosStopTcpServer(void *param);
void taosCleanUpTcpServer(void *param);
void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *fp, void *shandle);
void taosStopTcpClient(void *chandle);
void taosCleanUpTcpClient(void *chandle);
void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port);
......
src/rpc/inc/rpcUdp.h
浏览文件 @ a64fb809
......@@ -23,6 +23,7 @@ extern "C" {
#include "taosdef.h"
void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int, void *fp, void *shandle);
void taosStopUdpConnection(void *handle);
void taosCleanUpUdpConnection(void *handle);
int taosSendUdpData(uint32_t ip, uint16_t port, void *data, int dataLen, void *chandle);
void *taosOpenUdpConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port);
......
src/rpc/src/rpcMain.c
浏览文件 @ a64fb809
......@@ -153,6 +153,13 @@ void (*taosCleanUpConn[])(void *thandle) = {
taosCleanUpTcpClient
};
void (*taosStopConn[])(void *thandle) = {
taosStopUdpConnection,
taosStopUdpConnection,
taosStopTcpServer,
taosStopTcpClient,
};
int (*taosSendData[])(uint32_t ip, uint16_t port, void *data, int len, void *chandle) = {
taosSendUdpData,
taosSendUdpData,
......@@ -289,12 +296,18 @@ void *rpcOpen(const SRpcInit *pInit) {
void rpcClose(void *param) {
SRpcInfo *pRpc = (SRpcInfo *)param;
// stop connection to outside first
(*taosStopConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
(*taosStopConn[pRpc->connType])(pRpc->udphandle);
// close all connections
for (int i = 0; i < pRpc->sessions; ++i) {
if (pRpc->connList && pRpc->connList[i].user[0]) {
rpcCloseConn((void *)(pRpc->connList + i));
}
}
// clean up
(*taosCleanUpConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
(*taosCleanUpConn[pRpc->connType])(pRpc->udphandle);
......@@ -588,6 +601,7 @@ static void rpcReleaseConn(SRpcConn *pConn) {
pConn->inTranId = 0;
pConn->outTranId = 0;
pConn->secured = 0;
pConn->peerId = 0;
pConn->peerIp = 0;
pConn->peerPort = 0;
pConn->pReqMsg = NULL;
......@@ -627,6 +641,7 @@ static SRpcConn *rpcAllocateClientConn(SRpcInfo *pRpc) {
pConn->spi = pRpc->spi;
pConn->encrypt = pRpc->encrypt;
if (pConn->spi) memcpy(pConn->secret, pRpc->secret, TSDB_KEY_LEN);
tTrace("%s %p client connection is allocated", pRpc->label, pConn);
}
return pConn;
......@@ -681,6 +696,7 @@ static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, SRecvInfo *pRecv) {
}
taosHashPut(pRpc->hash, hashstr, size, (char *)&pConn, POINTER_BYTES);
tTrace("%s %p server connection is allocated", pRpc->label, pConn);
}
return pConn;
......@@ -948,11 +964,9 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
terrno = 0;
pConn = rpcProcessMsgHead(pRpc, pRecv);
if (pHead->msgType < TSDB_MSG_TYPE_CM_HEARTBEAT || (rpcDebugFlag & 16)) {
tTrace("%s %p %p, %s received from 0x%x:%hu, parse code:0x%x len:%d sig:0x%08x:0x%08x:%d code:0x%x",
tTrace("%s %p %p, %s received from 0x%x:%hu, parse code:0x%x len:%d sig:0x%08x:0x%08x:%d code:0x%x",
pRpc->label, pConn, (void *)pHead->ahandle, taosMsg[pHead->msgType], pRecv->ip, pRecv->port, terrno,
pRecv->msgLen, pHead->sourceId, pHead->destId, pHead->tranId, pHead->code);
}
int32_t code = terrno;
if (code != TSDB_CODE_RPC_ALREADY_PROCESSED) {
......@@ -1180,16 +1194,14 @@ static void rpcSendMsgToPeer(SRpcConn *pConn, void *msg, int msgLen) {
msgLen = rpcAddAuthPart(pConn, msg, msgLen);
if ( rpcIsReq(pHead->msgType)) {
if (pHead->msgType < TSDB_MSG_TYPE_CM_HEARTBEAT || (rpcDebugFlag & 16))
tTrace("%s, %s is sent to %s:%hu, len:%d sig:0x%08x:0x%08x:%d",
pConn->info, taosMsg[pHead->msgType], pConn->peerFqdn, pConn->peerPort,
msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
tTrace("%s, %s is sent to %s:%hu, len:%d sig:0x%08x:0x%08x:%d",
pConn->info, taosMsg[pHead->msgType], pConn->peerFqdn, pConn->peerPort,
msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
} else {
if (pHead->code == 0) pConn->secured = 1; // for success response, set link as secured
if (pHead->msgType < TSDB_MSG_TYPE_CM_HEARTBEAT || (rpcDebugFlag & 16))
tTrace("%s, %s is sent to 0x%x:%hu, code:0x%x len:%d sig:0x%08x:0x%08x:%d",
pConn->info, taosMsg[pHead->msgType], pConn->peerIp, pConn->peerPort,
htonl(pHead->code), msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
tTrace("%s, %s is sent to 0x%x:%hu, code:0x%x len:%d sig:0x%08x:0x%08x:%d",
pConn->info, taosMsg[pHead->msgType], pConn->peerIp, pConn->peerPort,
htonl(pHead->code), msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
}
//tTrace("connection type is: %d", pConn->connType);
......
src/rpc/src/rpcTcp.c
浏览文件 @ a64fb809
......@@ -190,22 +190,28 @@ static void taosStopTcpThread(SThreadObj* pThreadObj) {
}
}
void taosCleanUpTcpServer(void *handle) {
void taosStopTcpServer(void *handle) {
SServerObj *pServerObj = handle;
SThreadObj *pThreadObj;
if (pServerObj == NULL) return;
if(pServerObj->fd >=0) shutdown(pServerObj->fd, SHUT_RD);
if(pServerObj->thread) pthread_join(pServerObj->thread, NULL);
tTrace("%s TCP server is stopped", pServerObj->label);
}
void taosCleanUpTcpServer(void *handle) {
SServerObj *pServerObj = handle;
SThreadObj *pThreadObj;
if (pServerObj == NULL) return;
for (int i = 0; i < pServerObj->numOfThreads; ++i) {
pThreadObj = pServerObj->pThreadObj + i;
taosStopTcpThread(pThreadObj);
pthread_mutex_destroy(&(pThreadObj->mutex));
}
tTrace("TCP:%s, TCP server is cleaned up", pServerObj->label);
tTrace("%s TCP server is cleaned up", pServerObj->label);
tfree(pServerObj->pThreadObj);
tfree(pServerObj);
......@@ -226,7 +232,7 @@ static void *taosAcceptTcpConnection(void *arg) {
connFd = accept(pServerObj->fd, (struct sockaddr *)&caddr, &addrlen);
if (connFd == -1) {
if (errno == EINVAL) {
tTrace("%s TCP server socket was shutdown, exiting...", pServerObj->label);
tTrace("%s TCP server stop accepting new connections, exiting", pServerObj->label);
break;
}
......@@ -242,13 +248,13 @@ static void *taosAcceptTcpConnection(void *arg) {
SFdObj *pFdObj = taosMallocFdObj(pThreadObj, connFd);
if (pFdObj) {
pFdObj->ip = caddr.sin_addr.s_addr;
pFdObj->port = caddr.sin_port;
tTrace("%s new connection from %s:%hu, FD:%p, numOfFds:%d", pServerObj->label,
inet_ntoa(caddr.sin_addr), pFdObj->port, pFdObj, pThreadObj->numOfFds);
pFdObj->port = htons(caddr.sin_port);
tTrace("%s new TCP connection from %s:%hu, fd:%d FD:%p numOfFds:%d", pServerObj->label,
inet_ntoa(caddr.sin_addr), pFdObj->port, connFd, pFdObj, pThreadObj->numOfFds);
} else {
close(connFd);
tError("%s failed to malloc FdObj(%s) for connection from:%s:%hu", pServerObj->label, strerror(errno),
inet_ntoa(caddr.sin_addr), caddr.sin_port);
inet_ntoa(caddr.sin_addr), htons(caddr.sin_port));
}
// pick up next thread for next connection
......@@ -304,12 +310,19 @@ void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *
return pThreadObj;
}
void taosStopTcpClient(void *chandle) {
SThreadObj *pThreadObj = chandle;
if (pThreadObj == NULL) return;
tTrace ("%s TCP client is stopped", pThreadObj->label);
}
void taosCleanUpTcpClient(void *chandle) {
SThreadObj *pThreadObj = chandle;
if (pThreadObj == NULL) return;
taosStopTcpThread(pThreadObj);
tTrace ("%s, all connections are cleaned up", pThreadObj->label);
tTrace ("%s TCP client is cleaned up", pThreadObj->label);
tfree(pThreadObj);
}
......@@ -320,14 +333,22 @@ void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uin
int fd = taosOpenTcpClientSocket(ip, port, pThreadObj->ip);
if (fd < 0) return NULL;
struct sockaddr_in sin;
uint16_t localPort = 0;
unsigned int addrlen = sizeof(sin);
if (getsockname(fd, (struct sockaddr *)&sin, &addrlen) == 0 &&
sin.sin_family == AF_INET && addrlen == sizeof(sin)) {
localPort = (uint16_t)ntohs(sin.sin_port);
}
SFdObj *pFdObj = taosMallocFdObj(pThreadObj, fd);
if (pFdObj) {
pFdObj->thandle = thandle;
pFdObj->port = port;
pFdObj->ip = ip;
tTrace("%s %p, TCP connection to 0x%x:%hu is created, FD:%p numOfFds:%d",
pThreadObj->label, thandle, ip, port, pFdObj, pThreadObj->numOfFds);
tTrace("%s %p TCP connection to 0x%x:%hu is created, localPort:%hu FD:%p numOfFds:%d",
pThreadObj->label, thandle, ip, port, localPort, pFdObj, pThreadObj->numOfFds);
} else {
close(fd);
tError("%s failed to malloc client FdObj(%s)", pThreadObj->label, strerror(errno));
......@@ -340,7 +361,10 @@ void taosCloseTcpConnection(void *chandle) {
SFdObj *pFdObj = chandle;
if (pFdObj == NULL) return;
pFdObj->thandle = NULL;
SThreadObj *pThreadObj = pFdObj->pThreadObj;
tTrace("%s %p TCP connection will be closed, FD:%p", pThreadObj->label, pFdObj->thandle, pFdObj);
// pFdObj->thandle = NULL;
pFdObj->closedByApp = 1;
shutdown(pFdObj->fd, SHUT_WR);
}
......@@ -385,14 +409,14 @@ static int taosReadTcpData(SFdObj *pFdObj, SRecvInfo *pInfo) {
headLen = taosReadMsg(pFdObj->fd, &rpcHead, sizeof(SRpcHead));
if (headLen != sizeof(SRpcHead)) {
tTrace("%s %p, read error, headLen:%d", pThreadObj->label, pFdObj->thandle, headLen);
tTrace("%s %p read error, headLen:%d", pThreadObj->label, pFdObj->thandle, headLen);
return -1;
}
msgLen = (int32_t)htonl((uint32_t)rpcHead.msgLen);
buffer = malloc(msgLen + tsRpcOverhead);
if ( NULL == buffer) {
tError("%s %p, TCP malloc(size:%d) fail", pThreadObj->label, pFdObj->thandle, msgLen);
tError("%s %p TCP malloc(size:%d) fail", pThreadObj->label, pFdObj->thandle, msgLen);
return -1;
}
......@@ -401,8 +425,8 @@ static int taosReadTcpData(SFdObj *pFdObj, SRecvInfo *pInfo) {
retLen = taosReadMsg(pFdObj->fd, msg + headLen, leftLen);
if (leftLen != retLen) {
tError("%s %p, read error, leftLen:%d retLen:%d",
pThreadObj->label, pFdObj->thandle, leftLen, retLen);
tError("%s %p read error, leftLen:%d retLen:%d FD:%p",
pThreadObj->label, pFdObj->thandle, leftLen, retLen, pFdObj);
free(buffer);
return -1;
}
......@@ -446,19 +470,19 @@ static void *taosProcessTcpData(void *param) {
pFdObj = events[i].data.ptr;
if (events[i].events & EPOLLERR) {
tTrace("%s %p, error happened on FD", pThreadObj->label, pFdObj->thandle);
tTrace("%s %p FD:%p epoll errors", pThreadObj->label, pFdObj->thandle, pFdObj);
taosReportBrokenLink(pFdObj);
continue;
}
if (events[i].events & EPOLLRDHUP) {
tTrace("%s %p, FD RD hang up", pThreadObj->label, pFdObj->thandle);
tTrace("%s %p FD:%p RD hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
taosReportBrokenLink(pFdObj);
continue;
}
if (events[i].events & EPOLLHUP) {
tTrace("%s %p, FD hang up", pThreadObj->label, pFdObj->thandle);
tTrace("%s %p FD:%p hang up", pThreadObj->label, pFdObj->thandle, pFdObj);
taosReportBrokenLink(pFdObj);
continue;
}
......@@ -527,7 +551,7 @@ static void taosFreeFdObj(SFdObj *pFdObj) {
pThreadObj->numOfFds--;
if (pThreadObj->numOfFds < 0)
tError("%s %p, TCP thread:%d, number of FDs is negative!!!",
tError("%s %p TCP thread:%d, number of FDs is negative!!!",
pThreadObj->label, pFdObj->thandle, pThreadObj->threadId);
if (pFdObj->prev) {
......@@ -542,7 +566,7 @@ static void taosFreeFdObj(SFdObj *pFdObj) {
pthread_mutex_unlock(&pThreadObj->mutex);
tTrace("%s %p, FD:%p is cleaned, numOfFds:%d",
tTrace("%s %p TCP connection is closed, FD:%p numOfFds:%d",
pThreadObj->label, pFdObj->thandle, pFdObj, pThreadObj->numOfFds);
tfree(pFdObj);
......
src/rpc/src/rpcUdp.c
浏览文件 @ a64fb809
......@@ -30,7 +30,6 @@
#define RPC_MAX_UDP_SIZE 65480
typedef struct {
void *signature;
int index;
int fd;
uint16_t port; // peer port
......@@ -111,7 +110,6 @@ void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int threads
pConn->processData = fp;
pConn->index = i;
pConn->pSet = pSet;
pConn->signature = pConn;
int code = pthread_create(&pConn->thread, &thAttr, taosRecvUdpData, pConn);
if (code != 0) {
......@@ -132,7 +130,7 @@ void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int threads
return pSet;
}
void taosCleanUpUdpConnection(void *handle) {
void taosStopUdpConnection(void *handle) {
SUdpConnSet *pSet = (SUdpConnSet *)handle;
SUdpConn *pConn;
......@@ -140,8 +138,6 @@ void taosCleanUpUdpConnection(void *handle) {
for (int i = 0; i < pSet->threads; ++i) {
pConn = pSet->udpConn + i;
pConn->signature = NULL;
if (pConn->fd >=0) shutdown(pConn->fd, SHUT_RDWR);
if (pConn->fd >=0) taosCloseSocket(pConn->fd);
}
......@@ -150,9 +146,24 @@ void taosCleanUpUdpConnection(void *handle) {
pConn = pSet->udpConn + i;
if (pConn->thread) pthread_join(pConn->thread, NULL);
tfree(pConn->buffer);
tTrace("%s UDP thread is closed, inedx:%d", pConn->label, i);
// tTrace("%s UDP thread is closed, index:%d", pConn->label, i);
}
tTrace("%s UDP is stopped", pSet->label);
}
void taosCleanUpUdpConnection(void *handle) {
SUdpConnSet *pSet = (SUdpConnSet *)handle;
SUdpConn *pConn;
if (pSet == NULL) return;
for (int i = 0; i < pSet->threads; ++i) {
pConn = pSet->udpConn + i;
if (pConn->fd >=0) taosCloseSocket(pConn->fd);
}
tTrace("%s UDP is cleaned up", pSet->label);
tfree(pSet);
}
......@@ -164,7 +175,7 @@ void *taosOpenUdpConnection(void *shandle, void *thandle, uint32_t ip, uint16_t
SUdpConn *pConn = pSet->udpConn + pSet->index;
pConn->port = port;
tTrace("%s UDP connection is setup, ip:%x:%hu", pConn->label, ip, port);
tTrace("%s UDP connection is setup, ip:%x:%hu localPort:%hu", pConn->label, ip, port, pConn->localPort);
return pConn;
}
......@@ -185,7 +196,7 @@ static void *taosRecvUdpData(void *param) {
while (1) {
dataLen = recvfrom(pConn->fd, pConn->buffer, RPC_MAX_UDP_SIZE, 0, (struct sockaddr *)&sourceAdd, &addLen);
if(dataLen <= 0) {
tTrace("%s UDP socket was closed, exiting", pConn->label);
tTrace("%s UDP socket was closed, exiting(%s)", pConn->label, strerror(errno));
break;
}
......@@ -221,7 +232,7 @@ static void *taosRecvUdpData(void *param) {
int taosSendUdpData(uint32_t ip, uint16_t port, void *data, int dataLen, void *chandle) {
SUdpConn *pConn = (SUdpConn *)chandle;
if (pConn == NULL || pConn->signature != pConn) return -1;
if (pConn == NULL) return -1;
struct sockaddr_in destAdd;
memset(&destAdd, 0, sizeof(destAdd));
......
src/tsdb/src/tsdbBuffer.c
浏览文件 @ a64fb809
......@@ -124,6 +124,7 @@ SListNode *tsdbAllocBufBlockFromPool(STsdbRepo *pRepo) {
}
SListNode * pNode = tdListPopHead(pBufPool->bufBlockList);
ASSERT(pNode != NULL);
STsdbBufBlock *pBufBlock = NULL;
tdListNodeGetData(pBufPool->bufBlockList, pNode, (void *)(&pBufBlock));
......
src/tsdb/src/tsdbFile.c
浏览文件 @ a64fb809
......@@ -358,7 +358,9 @@ void tsdbRemoveFileGroup(STsdbRepo *pRepo, SFileGroup *pFGroup) {
ASSERT(pFileH->nFGroups >= 0);
for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
remove(fileGroup.files[type].fname);
if (remove(fileGroup.files[type].fname) < 0) {
tsdbError("vgId:%d failed to remove file %s", REPO_ID(pRepo), fileGroup.files[type].fname);
}
tsdbDestroyFile(&fileGroup.files[type]);
}
}
......
src/tsdb/src/tsdbMain.c
浏览文件 @ a64fb809
......@@ -213,10 +213,10 @@ uint32_t tsdbGetFileInfo(TSDB_REPO_T *repo, char *name, uint32_t *index, uint32_
SFileGroup *pFGroup =
taosbsearch(&fid, pFileH->pFGroup, pFileH->nFGroups, sizeof(SFileGroup), keyFGroupCompFunc, TD_GE);
if (pFGroup->fileId == fid) {
strcpy(fname, pFGroup->files[(*index) % 3].fname);
fname = strdup(pFGroup->files[(*index) % 3].fname);
} else {
if (pFGroup->fileId * 3 + 2 < eindex) {
strcpy(fname, pFGroup->files[0].fname);
fname = strdup(pFGroup->files[0].fname);
*index = pFGroup->fileId * 3;
} else {
tfree(sdup);
......@@ -237,12 +237,13 @@ uint32_t tsdbGetFileInfo(TSDB_REPO_T *repo, char *name, uint32_t *index, uint32_
}
SFile *pFile = &pFGroup->files[(*index) % 3];
strcpy(fname, pFile->fname);
fname = strdup(pFile->fname);
}
}
if (stat(fname, &fState) < 0) {
tfree(sdup);
tfree(fname);
return 0;
}
......@@ -566,7 +567,7 @@ static int32_t tsdbSaveConfig(char *rootDir, STsdbCfg *pCfg) {
_err:
tfree(fname);
if (fd > 0) close(fd);
if (fd >= 0) close(fd);
return -1;
}
......@@ -609,7 +610,7 @@ static int tsdbLoadConfig(char *rootDir, STsdbCfg *pCfg) {
_err:
tfree(fname);
if (fd > 0) close(fd);
if (fd >= 0) close(fd);
return -1;
}
......
src/tsdb/src/tsdbMemTable.c
浏览文件 @ a64fb809
......@@ -99,6 +99,7 @@ int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
if (tSkipListPut(pTableData->pData, pNode) == NULL) {
tsdbFreeBytes(pRepo, (void *)pNode, bytes);
} else {
if (TABLE_LASTKEY(pTable) < key) TABLE_LASTKEY(pTable) = key;
if (pMemTable->keyFirst > key) pMemTable->keyFirst = key;
if (pMemTable->keyLast < key) pMemTable->keyLast = key;
pMemTable->numOfRows++;
......@@ -222,11 +223,12 @@ int tsdbAsyncCommit(STsdbRepo *pRepo) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
pRepo->commit = 0;
}
ASSERT(pRepo->commit == 0);
if (pRepo->appH.notifyStatus) pRepo->appH.notifyStatus(pRepo->appH.appH, TSDB_STATUS_COMMIT_START);
if (pRepo->mem != NULL) {
if (pRepo->appH.notifyStatus) pRepo->appH.notifyStatus(pRepo->appH.appH, TSDB_STATUS_COMMIT_START);
if (tsdbLockRepo(pRepo) < 0) return -1;
pRepo->imem = pRepo->mem;
pRepo->mem = NULL;
......@@ -468,9 +470,6 @@ _err:
static void tsdbEndCommit(STsdbRepo *pRepo) {
ASSERT(pRepo->commit == 1);
tsdbLockRepo(pRepo);
pRepo->commit = 0;
tsdbUnlockRepo(pRepo);
if (pRepo->appH.notifyStatus) pRepo->appH.notifyStatus(pRepo->appH.appH, TSDB_STATUS_COMMIT_OVER);
}
......@@ -526,8 +525,6 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
goto _err;
}
free(dataDir);
// Open files for write/read
if (tsdbSetAndOpenHelperFile(pHelper, pGroup) < 0) {
tsdbError("vgId:%d failed to set helper file since %s", REPO_ID(pRepo), tstrerror(terrno));
......@@ -590,6 +587,7 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
goto _err;
}
tfree(dataDir);
tsdbCloseHelperFile(pHelper, 0);
pthread_rwlock_wrlock(&(pFileH->fhlock));
......@@ -601,7 +599,7 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
return 0;
_err:
// ASSERT(false);
tfree(dataDir);
tsdbCloseHelperFile(pHelper, 1);
return -1;
}
......
src/tsdb/src/tsdbMeta.c
浏览文件 @ a64fb809
......@@ -147,6 +147,7 @@ int tsdbDropTable(TSDB_REPO_T *repo, STableId tableId) {
tsdbInsertTableAct(pRepo, TSDB_DROP_META, buf, tTable);
tsdbRemoveTableFromMeta(pRepo, tTable, false, true);
}
tSkipListDestroyIter(pIter);
}
tsdbRemoveTableFromMeta(pRepo, pTable, true, true);
......@@ -269,7 +270,6 @@ STableCfg *tsdbCreateTableCfgFromMsg(SMDCreateTableMsg *pMsg) {
_err:
tdDestroyTSchemaBuilder(&schemaBuilder);
tsdbClearTableCfg(pCfg);
tfree(pCfg);
return NULL;
}
......@@ -308,6 +308,7 @@ int tsdbUpdateTagValue(TSDB_REPO_T *repo, SUpdateTableTagValMsg *pMsg) {
int32_t code = tsdbUpdateTable(pRepo, super, pTableCfg);
if (code != TSDB_CODE_SUCCESS) {
tsdbClearTableCfg(pTableCfg);
return code;
}
tsdbClearTableCfg(pTableCfg);
......
src/util/inc/tstoken.h
浏览文件 @ a64fb809
......@@ -24,14 +24,7 @@ extern "C" {
#include "tutil.h"
#include "ttokendef.h"
#define TK_SPACE 200
#define TK_COMMENT 201
#define TK_ILLEGAL 202
#define TK_HEX 203 // hex number 0x123
#define TK_OCT 204 // oct number
#define TK_BIN 205 // bin format data 0b111
#define TK_FILE 206
#define TK_QUESTION 207 // denoting the placeholder of "?",when invoking statement bind query
#define TSQL_TBNAME "TBNAME"
#define TSQL_TBNAME_L "tbname"
......
src/util/inc/tutil.h
浏览文件 @ a64fb809
......@@ -119,6 +119,8 @@ extern "C" {
uint32_t taosRand(void);
uint32_t trand(void);
size_t twcslen(const wchar_t *wcs);
int32_t strdequote(char *src);
......
src/util/src/tnote.c
浏览文件 @ a64fb809
......@@ -128,10 +128,12 @@ int taosOpenNewNote(taosNoteInfo * pNote)
bool taosCheckNoteIsOpen(char *noteName, taosNoteInfo * pNote)
{
/*
int exist = access(noteName, F_OK);
if (exist != 0) {
return false;
}
*/
int fd = open(noteName, O_WRONLY | O_CREAT, S_IRWXU | S_IRWXG | S_IRWXO);
if (fd < 0) {
......
src/util/src/tsocket.c
浏览文件 @ a64fb809
......@@ -222,9 +222,7 @@ int taosReadn(int fd, char *ptr, int nbytes) {
int taosOpenUdpSocket(uint32_t ip, uint16_t port) {
struct sockaddr_in localAddr;
int sockFd;
int ttl = 128;
int reuse, nocheck;
int bufSize = 8192000;
int bufSize = 1024000;
uTrace("open udp socket:0x%x:%hu", ip, port);
......@@ -238,31 +236,6 @@ int taosOpenUdpSocket(uint32_t ip, uint16_t port) {
return -1;
}
reuse = 1;
if (taosSetSockOpt(sockFd, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {
uError("setsockopt SO_REUSEADDR failed): %d (%s)", errno, strerror(errno));
close(sockFd);
return -1;
};
nocheck = 1;
if (taosSetSockOpt(sockFd, SOL_SOCKET, SO_NO_CHECK, (void *)&nocheck, sizeof(nocheck)) < 0) {
if (!taosSkipSocketCheck()) {
uError("setsockopt SO_NO_CHECK failed: %d (%s)", errno, strerror(errno));
close(sockFd);
return -1;
} else {
uPrint("Skipping setsockopt SO_NO_CHECK error: %d (%s)", errno, strerror(errno));
}
}
ttl = 128;
if (taosSetSockOpt(sockFd, IPPROTO_IP, IP_TTL, (void *)&ttl, sizeof(ttl)) < 0) {
uError("setsockopt IP_TTL failed: %d (%s)", errno, strerror(errno));
close(sockFd);
return -1;
}
if (taosSetSockOpt(sockFd, SOL_SOCKET, SO_SNDBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the send buffer size for UDP socket\n");
close(sockFd);
......
src/util/src/tutil.c
浏览文件 @ a64fb809
......@@ -55,6 +55,25 @@ uint32_t taosRand(void)
*/
return rand();
}
uint32_t trand(void)
{
int fd;
int seed;
fd = open("/dev/urandom", 0);
if (fd < 0) {
seed = time(0);
} else {
int len = read(fd, &seed, sizeof(seed));
if (len < 0) {
seed = time(0);
}
close(fd);
}
return (uint32_t)seed;
}
#endif
size_t twcslen(const wchar_t *wcs) {
......
tests/pytest/table/boundary.py
浏览文件 @ a64fb809
......@@ -141,7 +141,7 @@ class TDTestCase:
tdSql.prepare()
# 8 bytes for timestamp
maxRowSize = 65535 - 8
maxRowSize = self.getLimitFromSourceCode('TSDB_MAX_BYTES_PER_ROW') - 8
maxCols = self.getLimitFromSourceCode('TSDB_MAX_COLUMNS') - 1
# for binary cols, 2 bytes are used for length
......
tests/script/general/stable/refcount.sim 0 → 100644
浏览文件 @ a64fb809
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c walLevel -v 0
system sh/cfg.sh -n dnode1 -c numOfTotalVnodes -v 4
system sh/cfg.sh -n dnode1 -c maxtablesPerVnode -v 4
system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
print =============== step1
sql create database d1;
sql use d1;
sql create table d1.t1 (ts timestamp, i int);
sql create table d1.t2 (ts timestamp, i int);
sql create table d1.t3 (ts timestamp, i int);
sql insert into d1.t1 values(now, 1);
sql insert into d1.t2 values(now, 1);
sql drop table d1.t1;
sql drop database d1;
sql show databases;
if $rows != 0 then
return -1
endi
print =============== step2
sql create database d2;
sql use d2;
sql create table d2.t1 (ts timestamp, i int);
sql create table d2.t2 (ts timestamp, i int);
sql create table d2.t3 (ts timestamp, i int);
sql insert into d2.t1 values(now, 1);
sql insert into d2.t2 values(now, 1);
sql drop table d2.t1;
sql drop table d2.t2;
sql drop table d2.t3;
sql show d2.tables;
if $rows != 0 then
return -1
endi
sql show d2.vgroups;
if $rows != 0 then
return -1
endi
sql drop database d2;
sql show databases;
if $rows != 0 then
return -1
endi
print =============== step3
sql create database d3;
sql use d3;
sql create table d3.st (ts timestamp, i int) tags (j int);
sql create table d3.t1 using d3.st tags(1);
sql create table d3.t2 using d3.st tags(1);
sql create table d3.t3 using d3.st tags(1);
sql insert into d3.t1 values(now, 1);
sql drop table d3.t1;
sql drop table d3.t2;
sql drop table d3.t3;
sql show d3.tables;
if $rows != 0 then
return -1
endi
sql show d3.vgroups;
if $rows != 0 then
return -1
endi
sql drop database d3;
sql show databases;
if $rows != 0 then
return -1
endi
print =============== step4
sql create database d4;
sql use d4;
sql create table d4.st (ts timestamp, i int) tags (j int);
sql create table d4.t1 using d4.st tags(1);
sql create table d4.t2 using d4.st tags(1);
sql create table d4.t3 using d4.st tags(1);
sql insert into d4.t1 values(now, 1);
sql drop table d4.t1;
sql drop table d4.st;
sql show d4.tables;
if $rows != 0 then
return -1
endi
sql show d4.stables;
if $rows != 0 then
return -1
endi
sql drop database d4;
sql show databases;
if $rows != 0 then
return -1
endi
print =============== step5
sql create database d5;
sql use d5;
sql create table d5.st (ts timestamp, i int) tags (j int);
sql create table d5.t1 using d5.st tags(1);
sql create table d5.t2 using d5.st tags(1);
sql create table d5.t3 using d5.st tags(1);
sql insert into d5.t1 values(now, 1);
sql drop table d5.t1;
sql drop database d5;
sql show databases;
if $rows != 0 then
return -1
endi
print =============== step6
system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file
tests/script/jenkins/unique.txt
浏览文件 @ a64fb809
......@@ -35,6 +35,7 @@ cd ../../../debug; make
./test.sh -f unique/db/replica_reduce31.sim
./test.sh -f unique/db/replica_part.sim
./test.sh -f unique/dnode/alternativeRole.sim
./test.sh -f unique/dnode/balance1.sim
./test.sh -f unique/dnode/balance2.sim
./test.sh -f unique/dnode/balance3.sim
......@@ -75,3 +76,59 @@ cd ../../../debug; make
./test.sh -f unique/vnode/replica3_basic.sim
./test.sh -f unique/vnode/replica3_repeat.sim
./test.sh -f unique/vnode/replica3_vgroup.sim
./test.sh -f general/stream/metrics_1.sim
./test.sh -f general/stream/metrics_del.sim
./test.sh -f general/stream/metrics_n.sim
./test.sh -f general/stream/metrics_replica1_vnoden.sim
#./test.sh -f general/stream/new_stream.sim
./test.sh -f general/stream/restart_stream.sim
./test.sh -f general/stream/stream_1.sim
./test.sh -f general/stream/stream_2.sim
./test.sh -f general/stream/stream_3.sim
./test.sh -f general/stream/stream_restart.sim
./test.sh -f general/stream/table_1.sim
./test.sh -f general/stream/table_del.sim
./test.sh -f general/stream/table_n.sim
./test.sh -f general/stream/table_replica1_vnoden.sim
./test.sh -f unique/arbitrator/check_cluster_cfg_para.sim
./test.sh -f unique/arbitrator/dn2_mn1_cache_file_sync.sim
./test.sh -f unique/arbitrator/dn3_mn1_full_createTableFail.sim
./test.sh -f unique/arbitrator/dn3_mn1_full_dropDnodeFail.sim
./test.sh -f unique/arbitrator/dn3_mn1_multiCreateDropTable.sim
./test.sh -f unique/arbitrator/dn3_mn1_nw_disable_timeout_autoDropDnode.sim
./test.sh -f unique/arbitrator/dn3_mn1_replica2_wal1_AddDelDnode.sim
./test.sh -f unique/arbitrator/dn3_mn1_replica_change_dropDnod.sim
./test.sh -f unique/arbitrator/dn3_mn1_replica_change.sim
./test.sh -f unique/arbitrator/dn3_mn1_stopDnode_timeout.sim
./test.sh -f unique/arbitrator/dn3_mn1_vnode_change.sim
./test.sh -f unique/arbitrator/dn3_mn1_vnode_corruptFile_offline.sim
./test.sh -f unique/arbitrator/dn3_mn1_vnode_corruptFile_online.sim
./test.sh -f unique/arbitrator/dn3_mn1_vnode_noCorruptFile_offline.sim
./test.sh -f unique/arbitrator/dn3_mn1_vnode_delDir.sim
./test.sh -f unique/arbitrator/dn3_mn1_r2_vnode_delDir.sim
./test.sh -f unique/arbitrator/dn3_mn1_r3_vnode_delDir.sim
./test.sh -f unique/arbitrator/dn3_mn1_vnode_nomaster.sim
./test.sh -f unique/arbitrator/dn3_mn2_killDnode.sim
./test.sh -f unique/arbitrator/insert_duplicationTs.sim
./test.sh -f unique/arbitrator/offline_replica2_alterTable_online.sim
./test.sh -f unique/arbitrator/offline_replica2_alterTag_online.sim
./test.sh -f unique/arbitrator/offline_replica2_createTable_online.sim
./test.sh -f unique/arbitrator/offline_replica2_dropDb_online.sim
./test.sh -f unique/arbitrator/offline_replica2_dropTable_online.sim
./test.sh -f unique/arbitrator/offline_replica3_alterTable_online.sim
./test.sh -f unique/arbitrator/offline_replica3_alterTag_online.sim
./test.sh -f unique/arbitrator/offline_replica3_createTable_online.sim
./test.sh -f unique/arbitrator/offline_replica3_dropDb_online.sim
./test.sh -f unique/arbitrator/offline_replica3_dropTable_online.sim
./test.sh -f unique/arbitrator/replica_changeWithArbitrator.sim
./test.sh -f unique/arbitrator/sync_replica2_alterTable_add.sim
./test.sh -f unique/arbitrator/sync_replica2_alterTable_drop.sim
./test.sh -f unique/arbitrator/sync_replica2_dropDb.sim
./test.sh -f unique/arbitrator/sync_replica2_dropTable.sim
./test.sh -f unique/arbitrator/sync_replica3_alterTable_add.sim
./test.sh -f unique/arbitrator/sync_replica3_alterTable_drop.sim
./test.sh -f unique/arbitrator/sync_replica3_dropDb.sim
./test.sh -f unique/arbitrator/sync_replica3_dropTable.sim
tests/script/loop.sh 0 → 100755
浏览文件 @ a64fb809
#!/bin/bash
##################################################
#
# Do simulation test
#
##################################################
set -e
#set -x
CMD_NAME=
LOOP_TIMES=5
while getopts "f:t:" arg
do
case $arg in
f)
CMD_NAME=$OPTARG
;;
t)
LOOP_TIMES=$OPTARG
;;
?)
echo "unknow argument"
;;
esac
done
echo LOOP_TIMES ${LOOP_TIMES}
echo CMD_NAME ${CMD_NAME}
for ((i=0; i<$LOOP_TIMES; i++ ))
do
echo loop $i
echo cmd $CMD_NAME
$CMD_NAME
sleep 2
done
tests/script/unique/cluster/balance2.sim
浏览文件 @ a64fb809
......@@ -328,7 +328,7 @@ $x = 0
show6:
$x = $x + 1
sleep 2000
if $x == 30 then
if $x == 10 then
return -1
endi
sql show dnodes -x show6
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或