提交 335d42cf 编写于 作者: S slguan

fix the issue #443

上级 b806c458
...@@ -66,7 +66,7 @@ int main(int argc, char *argv[]) { ...@@ -66,7 +66,7 @@ int main(int argc, char *argv[]) {
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
} else if (strcmp(argv[i], "-V") == 0) { } else if (strcmp(argv[i], "-V") == 0) {
printf("%s %s\n", version, compatible_version); printf("version: %s compatible_version: %s\n", version, compatible_version);
printf("gitinfo: %s\n", gitinfo); printf("gitinfo: %s\n", gitinfo);
printf("buildinfo: %s\n", buildinfo); printf("buildinfo: %s\n", buildinfo);
return 0; return 0;
......
...@@ -29,6 +29,7 @@ char *mgmtBuildCreateMeterIe(STabObj *pMeter, char *pMsg, int vnode); ...@@ -29,6 +29,7 @@ char *mgmtBuildCreateMeterIe(STabObj *pMeter, char *pMsg, int vnode);
void vnodeProcessMsgFromMgmt(SSchedMsg *smsg); void vnodeProcessMsgFromMgmt(SSchedMsg *smsg);
void *rpcQhandle; void *rpcQhandle;
extern void *dmQhandle;
int mgmtSendMsgToDnode(char *msg) { int mgmtSendMsgToDnode(char *msg) {
mTrace("msg:%s is sent to dnode", taosMsg[*msg]); mTrace("msg:%s is sent to dnode", taosMsg[*msg]);
...@@ -38,7 +39,7 @@ int mgmtSendMsgToDnode(char *msg) { ...@@ -38,7 +39,7 @@ int mgmtSendMsgToDnode(char *msg) {
schedMsg.msg = msg; schedMsg.msg = msg;
schedMsg.ahandle = NULL; schedMsg.ahandle = NULL;
schedMsg.thandle = NULL; schedMsg.thandle = NULL;
taosScheduleTask(rpcQhandle, &schedMsg); taosScheduleTask(dmQhandle, &schedMsg);
return 0; return 0;
} }
......
...@@ -144,6 +144,7 @@ size_t vnodeRestoreDataFromLog(int vnode, char *fileName, uint64_t *firstV) { ...@@ -144,6 +144,7 @@ size_t vnodeRestoreDataFromLog(int vnode, char *fileName, uint64_t *firstV) {
goto _error; goto _error;
} }
TSKEY now = taosGetTimestamp(pVnode->cfg.precision);
SCommitHead head; SCommitHead head;
int simpleCheck = 0; int simpleCheck = 0;
while (1) { while (1) {
...@@ -180,7 +181,7 @@ size_t vnodeRestoreDataFromLog(int vnode, char *fileName, uint64_t *firstV) { ...@@ -180,7 +181,7 @@ size_t vnodeRestoreDataFromLog(int vnode, char *fileName, uint64_t *firstV) {
int32_t numOfPoints = 0; int32_t numOfPoints = 0;
(*vnodeProcessAction[head.action])(pObj, cont, head.contLen, TSDB_DATA_SOURCE_LOG, NULL, head.sversion, (*vnodeProcessAction[head.action])(pObj, cont, head.contLen, TSDB_DATA_SOURCE_LOG, NULL, head.sversion,
&numOfPoints); &numOfPoints, now);
actions++; actions++;
} else { } else {
break; break;
......
...@@ -35,7 +35,7 @@ int tsMeterSizeOnFile; ...@@ -35,7 +35,7 @@ int tsMeterSizeOnFile;
void vnodeUpdateMeter(void *param, void *tmdId); void vnodeUpdateMeter(void *param, void *tmdId);
void vnodeRecoverMeterObjectFile(int vnode); void vnodeRecoverMeterObjectFile(int vnode);
int (*vnodeProcessAction[])(SMeterObj *, char *, int, char, void *, int, int *) = {vnodeInsertPoints, int (*vnodeProcessAction[])(SMeterObj *, char *, int, char, void *, int, int *, TSKEY) = {vnodeInsertPoints,
vnodeImportPoints}; vnodeImportPoints};
void vnodeFreeMeterObj(SMeterObj *pObj) { void vnodeFreeMeterObj(SMeterObj *pObj) {
...@@ -506,7 +506,7 @@ int vnodeRemoveMeterObj(int vnode, int sid) { ...@@ -506,7 +506,7 @@ int vnodeRemoveMeterObj(int vnode, int sid) {
} }
int vnodeInsertPoints(SMeterObj *pObj, char *cont, int contLen, char source, void *param, int sversion, int vnodeInsertPoints(SMeterObj *pObj, char *cont, int contLen, char source, void *param, int sversion,
int *numOfInsertPoints) { int *numOfInsertPoints, TSKEY now) {
int expectedLen, i; int expectedLen, i;
short numOfPoints; short numOfPoints;
SSubmitMsg *pSubmit = (SSubmitMsg *)cont; SSubmitMsg *pSubmit = (SSubmitMsg *)cont;
...@@ -528,7 +528,7 @@ int vnodeInsertPoints(SMeterObj *pObj, char *cont, int contLen, char source, voi ...@@ -528,7 +528,7 @@ int vnodeInsertPoints(SMeterObj *pObj, char *cont, int contLen, char source, voi
// to guarantee time stamp is the same for all vnodes // to guarantee time stamp is the same for all vnodes
pData = pSubmit->payLoad; pData = pSubmit->payLoad;
tsKey = taosGetTimestamp(pVnode->cfg.precision); tsKey = now;
cfile = tsKey/pVnode->cfg.daysPerFile/tsMsPerDay[pVnode->cfg.precision]; cfile = tsKey/pVnode->cfg.daysPerFile/tsMsPerDay[pVnode->cfg.precision];
if (*((TSKEY *)pData) == 0) { if (*((TSKEY *)pData) == 0) {
for (i = 0; i < numOfPoints; ++i) { for (i = 0; i < numOfPoints; ++i) {
......
...@@ -484,6 +484,7 @@ int vnodeProcessShellSubmitRequest(char *pMsg, int msgLen, SShellObj *pObj) { ...@@ -484,6 +484,7 @@ int vnodeProcessShellSubmitRequest(char *pMsg, int msgLen, SShellObj *pObj) {
int32_t numOfPoints = 0; int32_t numOfPoints = 0;
int32_t numOfTotalPoints = 0; int32_t numOfTotalPoints = 0;
TSKEY now = taosGetTimestamp(pVnode->cfg.precision);
for (int32_t i = 0; i < pSubmit->numOfSid; ++i) { for (int32_t i = 0; i < pSubmit->numOfSid; ++i) {
numOfPoints = 0; numOfPoints = 0;
...@@ -523,11 +524,11 @@ int vnodeProcessShellSubmitRequest(char *pMsg, int msgLen, SShellObj *pObj) { ...@@ -523,11 +524,11 @@ int vnodeProcessShellSubmitRequest(char *pMsg, int msgLen, SShellObj *pObj) {
// meter status is ready for insert/import // meter status is ready for insert/import
if (pSubmit->import) { if (pSubmit->import) {
code = vnodeImportPoints(pMeterObj, (char *) &(pBlocks->numOfRows), subMsgLen, TSDB_DATA_SOURCE_SHELL, pObj, code = vnodeImportPoints(pMeterObj, (char *) &(pBlocks->numOfRows), subMsgLen, TSDB_DATA_SOURCE_SHELL, pObj,
sversion, &numOfPoints); sversion, &numOfPoints, now);
vnodeClearMeterState(pMeterObj, TSDB_METER_STATE_IMPORTING); vnodeClearMeterState(pMeterObj, TSDB_METER_STATE_IMPORTING);
} else { } else {
code = vnodeInsertPoints(pMeterObj, (char *) &(pBlocks->numOfRows), subMsgLen, TSDB_DATA_SOURCE_SHELL, NULL, code = vnodeInsertPoints(pMeterObj, (char *) &(pBlocks->numOfRows), subMsgLen, TSDB_DATA_SOURCE_SHELL, NULL,
sversion, &numOfPoints); sversion, &numOfPoints, now);
vnodeClearMeterState(pMeterObj, TSDB_METER_STATE_INSERT); vnodeClearMeterState(pMeterObj, TSDB_METER_STATE_INSERT);
} }
......
...@@ -57,7 +57,7 @@ void vnodeProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) { ...@@ -57,7 +57,7 @@ void vnodeProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
int32_t state = vnodeSetMeterState(pObj, TSDB_METER_STATE_INSERT); int32_t state = vnodeSetMeterState(pObj, TSDB_METER_STATE_INSERT);
if (state == TSDB_METER_STATE_READY) { if (state == TSDB_METER_STATE_READY) {
vnodeInsertPoints(pObj, (char *)pMsg, contLen, TSDB_DATA_SOURCE_SHELL, NULL, pObj->sversion, &numOfPoints); vnodeInsertPoints(pObj, (char *)pMsg, contLen, TSDB_DATA_SOURCE_SHELL, NULL, pObj->sversion, &numOfPoints, taosGetTimestamp(vnodeList[pObj->vnode].cfg.precision));
vnodeClearMeterState(pObj, TSDB_METER_STATE_INSERT); vnodeClearMeterState(pObj, TSDB_METER_STATE_INSERT);
} else { } else {
dError("vid:%d sid:%d id:%s, failed to insert continuous query results, state:%d", pObj->vnode, pObj->sid, dError("vid:%d sid:%d id:%s, failed to insert continuous query results, state:%d", pObj->vnode, pObj->sid,
......
...@@ -41,9 +41,11 @@ int vnodeInitSystem() { ...@@ -41,9 +41,11 @@ int vnodeInitSystem() {
if (numOfThreads < 1) numOfThreads = 1; if (numOfThreads < 1) numOfThreads = 1;
queryQhandle = taosInitScheduler(tsNumOfVnodesPerCore * tsNumOfCores * tsSessionsPerVnode, numOfThreads, "query"); queryQhandle = taosInitScheduler(tsNumOfVnodesPerCore * tsNumOfCores * tsSessionsPerVnode, numOfThreads, "query");
// numOfThreads = (1.0 - tsRatioOfQueryThreads) * tsNumOfCores * tsNumOfThreadsPerCore / 2.0; numOfThreads = (1.0 - tsRatioOfQueryThreads) * tsNumOfCores * tsNumOfThreadsPerCore / 2.0;
// if (numOfThreads < 1) numOfThreads = 1; if (numOfThreads < 1) numOfThreads = 1;
rpcQhandle = taosInitScheduler(tsNumOfVnodesPerCore * tsNumOfCores * tsSessionsPerVnode, 1, "dnode"); rpcQhandle = taosInitScheduler(tsNumOfVnodesPerCore * tsNumOfCores * tsSessionsPerVnode, numOfThreads, "dnode");
dmQhandle = taosInitScheduler(tsSessionsPerVnode, 1, "mgmt");
vnodeTmrCtrl = taosTmrInit(tsSessionsPerVnode + 1000, 200, 60000, "DND-vnode"); vnodeTmrCtrl = taosTmrInit(tsSessionsPerVnode + 1000, 200, 60000, "DND-vnode");
if (vnodeTmrCtrl == NULL) { if (vnodeTmrCtrl == NULL) {
...@@ -70,11 +72,3 @@ int vnodeInitSystem() { ...@@ -70,11 +72,3 @@ int vnodeInitSystem() {
return 0; return 0;
} }
void vnodeInitQHandle() {
// int numOfThreads = (1.0 - tsRatioOfQueryThreads) * tsNumOfCores * tsNumOfThreadsPerCore / 2.0;
// if (numOfThreads < 1) numOfThreads = 1;
rpcQhandle = taosInitScheduler(tsNumOfVnodesPerCore * tsNumOfCores * tsSessionsPerVnode, 1, "dnode");
dmQhandle = taosInitScheduler(tsSessionsPerVnode, 1, "mgmt");
}
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册