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

Merge pull request #25646 from taosdata/fix/3_liaohj 

fix(stream): set global close flag in the streamMeta.
This commit is contained in:
Haojun Liao 2024-05-06 22:19:23 +08:00 committed by GitHub
parent 12f38f7685 18c66a5491
commit 3b43b8f8c7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 505 additions and 564 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/common/rsync.h
View File

@ -13,9 +13,9 @@ extern "C" {
void stopRsync(); void stopRsync();
void startRsync(); void startRsync();
int uploadRsync(char* id, char* path); int uploadRsync(const char* id, const char* path);
int downloadRsync(char* id, char* path); int downloadRsync(const char* id, const char* path);
int deleteRsync(char* id); int deleteRsync(const char* id);
#ifdef __cplusplus #ifdef __cplusplus
} }

1
include/dnode/vnode/tqCommon.h
View File

@ -33,7 +33,6 @@ int32_t tqStreamTaskProcessDeployReq(SStreamMeta* pMeta, SMsgCb* cb, int64_t sve
bool isLeader, bool restored); bool isLeader, bool restored);
int32_t tqStreamTaskProcessDropReq(SStreamMeta* pMeta, char* msg, int32_t msgLen); int32_t tqStreamTaskProcessDropReq(SStreamMeta* pMeta, char* msg, int32_t msgLen);
int32_t tqStreamTaskProcessRunReq(SStreamMeta* pMeta, SRpcMsg* pMsg, bool isLeader); int32_t tqStreamTaskProcessRunReq(SStreamMeta* pMeta, SRpcMsg* pMsg, bool isLeader);
int32_t tqStreamTaskResetStatus(SStreamMeta* pMeta);
int32_t tqStartTaskCompleteCallback(SStreamMeta* pMeta); int32_t tqStartTaskCompleteCallback(SStreamMeta* pMeta);
int32_t tqStreamTasksGetTotalNum(SStreamMeta* pMeta); int32_t tqStreamTasksGetTotalNum(SStreamMeta* pMeta);
int32_t tqStreamTaskProcessTaskResetReq(SStreamMeta* pMeta, SRpcMsg* pMsg); int32_t tqStreamTaskProcessTaskResetReq(SStreamMeta* pMeta, SRpcMsg* pMsg);

16
include/libs/stream/tstream.h
View File

@ -530,6 +530,7 @@ typedef struct SStreamMeta {
int32_t vgId; int32_t vgId;
int64_t stage; int64_t stage;
int32_t role; int32_t role;
bool closeFlag;
bool sendMsgBeforeClosing; // send hb to mnode before close all tasks when switch to follower. bool sendMsgBeforeClosing; // send hb to mnode before close all tasks when switch to follower.
STaskStartInfo startInfo; STaskStartInfo startInfo;
TdThreadRwlock lock; TdThreadRwlock lock;
@ -553,6 +554,12 @@ typedef struct SStreamMeta {
void* bkdChkptMgt; void* bkdChkptMgt;
} SStreamMeta; } SStreamMeta;
typedef struct STaskUpdateEntry {
int64_t streamId;
int32_t taskId;
int32_t transId;
} STaskUpdateEntry;
int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo); int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo);
int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamChildEpInfo* pInfo); int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamChildEpInfo* pInfo);
@ -792,7 +799,12 @@ SStreamChildEpInfo* streamTaskGetUpstreamTaskEpInfo(SStreamTask* pTask, int32_t
void streamTaskInputFail(SStreamTask* pTask); void streamTaskInputFail(SStreamTask* pTask);
int32_t streamExecTask(SStreamTask* pTask); int32_t streamExecTask(SStreamTask* pTask);
int32_t streamResumeTask(SStreamTask* pTask); int32_t streamResumeTask(SStreamTask* pTask);
int32_t streamSchedExec(SStreamTask* pTask); int32_t streamTrySchedExec(SStreamTask* pTask);
int32_t streamTaskSchedTask(SMsgCb* pMsgCb, int32_t vgId, int64_t streamId, int32_t taskId, int32_t execType);
int32_t streamTaskResumeInFuture(SStreamTask* pTask);
void streamTaskClearSchedIdleInfo(SStreamTask* pTask);
void streamTaskSetIdleInfo(SStreamTask* pTask, int32_t idleTime);
bool streamTaskShouldStop(const SStreamTask* pStatus); bool streamTaskShouldStop(const SStreamTask* pStatus);
bool streamTaskShouldPause(const SStreamTask* pStatus); bool streamTaskShouldPause(const SStreamTask* pStatus);
bool streamTaskIsIdle(const SStreamTask* pTask); bool streamTaskIsIdle(const SStreamTask* pTask);
@ -888,6 +900,8 @@ int32_t streamMetaAddTaskLaunchResult(SStreamMeta* pMeta, int64_t streamId,
int64_t endTs, bool ready); int64_t endTs, bool ready);
int32_t streamMetaResetTaskStatus(SStreamMeta* pMeta); int32_t streamMetaResetTaskStatus(SStreamMeta* pMeta);
int32_t streamMetaAddFailedTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId); int32_t streamMetaAddFailedTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId);
void streamMetaAddIntoUpdateTaskList(SStreamMeta* pMeta, SStreamTask* pTask, SStreamTask* pHTask, int32_t transId,
int64_t startTs);
void streamMetaRLock(SStreamMeta* pMeta); void streamMetaRLock(SStreamMeta* pMeta);
void streamMetaRUnLock(SStreamMeta* pMeta); void streamMetaRUnLock(SStreamMeta* pMeta);

14
source/common/src/rsync.c
View File

@ -9,7 +9,7 @@
#define ERRNO_ERR_DATA errno,strerror(errno) #define ERRNO_ERR_DATA errno,strerror(errno)
// deleteRsync function produce empty directories, traverse base directory to remove them // deleteRsync function produce empty directories, traverse base directory to remove them
static void removeEmptyDir(){ static void removeEmptyDir() {
TdDirPtr pDir = taosOpenDir(tsCheckpointBackupDir); TdDirPtr pDir = taosOpenDir(tsCheckpointBackupDir);
if (pDir == NULL) return; if (pDir == NULL) return;
@ -53,7 +53,7 @@ static void changeDirFromWindowsToLinux(char* from, char* to){
} }
#endif #endif
static int generateConfigFile(char* confDir){ static int generateConfigFile(char* confDir) {
TdFilePtr pFile = taosOpenFile(confDir, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); TdFilePtr pFile = taosOpenFile(confDir, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) { if (pFile == NULL) {
uError("[rsync] open conf file error, dir:%s,"ERRNO_ERR_FORMAT, confDir, ERRNO_ERR_DATA); uError("[rsync] open conf file error, dir:%s,"ERRNO_ERR_FORMAT, confDir, ERRNO_ERR_DATA);
@ -111,7 +111,7 @@ static int execCommand(char* command){
return code; return code;
} }
void stopRsync(){ void stopRsync() {
int code = int code =
#ifdef WINDOWS #ifdef WINDOWS
system("taskkill /f /im rsync.exe"); system("taskkill /f /im rsync.exe");
@ -125,7 +125,7 @@ void stopRsync(){
uDebug("[rsync] stop rsync server successful"); uDebug("[rsync] stop rsync server successful");
} }
void startRsync(){ void startRsync() {
if(taosMulMkDir(tsCheckpointBackupDir) != 0){ if(taosMulMkDir(tsCheckpointBackupDir) != 0){
uError("[rsync] build checkpoint backup dir failed, dir:%s,"ERRNO_ERR_FORMAT, tsCheckpointBackupDir, ERRNO_ERR_DATA); uError("[rsync] build checkpoint backup dir failed, dir:%s,"ERRNO_ERR_FORMAT, tsCheckpointBackupDir, ERRNO_ERR_DATA);
return; return;
@ -151,7 +151,7 @@ void startRsync(){
uDebug("[rsync] start server successful"); uDebug("[rsync] start server successful");
} }
int uploadRsync(char* id, char* path){ int uploadRsync(const char* id, const char* path) {
#ifdef WINDOWS #ifdef WINDOWS
char pathTransform[PATH_MAX] = {0}; char pathTransform[PATH_MAX] = {0};
changeDirFromWindowsToLinux(path, pathTransform); changeDirFromWindowsToLinux(path, pathTransform);
@ -188,7 +188,7 @@ int uploadRsync(char* id, char* path){
return 0; return 0;
} }
int downloadRsync(char* id, char* path){ int downloadRsync(const char* id, const char* path) {
#ifdef WINDOWS #ifdef WINDOWS
char pathTransform[PATH_MAX] = {0}; char pathTransform[PATH_MAX] = {0};
changeDirFromWindowsToLinux(path, pathTransform); changeDirFromWindowsToLinux(path, pathTransform);
@ -212,7 +212,7 @@ int downloadRsync(char* id, char* path){
return 0; return 0;
} }
int deleteRsync(char* id){ int deleteRsync(const char* id) {
char* tmp = "./tmp_empty/"; char* tmp = "./tmp_empty/";
int code = taosMkDir(tmp); int code = taosMkDir(tmp);
if(code != 0){ if(code != 0){

64
source/dnode/vnode/src/tq/tqStreamTask.c
View File

@ -62,29 +62,14 @@ typedef struct SBuildScanWalMsgParam {
} SBuildScanWalMsgParam; } SBuildScanWalMsgParam;
static void doStartScanWal(void* param, void* tmrId) { static void doStartScanWal(void* param, void* tmrId) {
SBuildScanWalMsgParam* pParam = (SBuildScanWalMsgParam*) param; SBuildScanWalMsgParam* pParam = (SBuildScanWalMsgParam*)param;
int32_t vgId = pParam->pTq->pStreamMeta->vgId;
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));
if (pRunReq == NULL) {
taosMemoryFree(pParam);
terrno = TSDB_CODE_OUT_OF_MEMORY;
tqError("vgId:%d failed to create msg to start wal scanning to launch stream tasks, code:%s", vgId, terrstr());
return;
}
STQ* pTq = pParam->pTq;
int32_t vgId = pTq->pStreamMeta->vgId;
tqDebug("vgId:%d create msg to start wal scan, numOfTasks:%d, vnd restored:%d", vgId, pParam->numOfTasks, tqDebug("vgId:%d create msg to start wal scan, numOfTasks:%d, vnd restored:%d", vgId, pParam->numOfTasks,
pParam->pTq->pVnode->restored); pTq->pVnode->restored);
pRunReq->head.vgId = vgId;
pRunReq->streamId = 0;
pRunReq->taskId = 0;
pRunReq->reqType = STREAM_EXEC_T_EXTRACT_WAL_DATA;
SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
tmsgPutToQueue(&pParam->pTq->pVnode->msgCb, STREAM_QUEUE, &msg);
/*int32_t code = */ streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_EXTRACT_WAL_DATA);
taosMemoryFree(pParam); taosMemoryFree(pParam);
} }
@ -149,50 +134,19 @@ int32_t tqScanWalAsync(STQ* pTq, bool ckPause) {
return 0; return 0;
} }
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));
if (pRunReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tqError("vgId:%d failed to create msg to start wal scanning to launch stream tasks, code:%s", vgId, terrstr());
streamMetaWUnLock(pMeta);
return -1;
}
tqDebug("vgId:%d create msg to start wal scan to launch stream tasks, numOfTasks:%d, vnd restored:%d", vgId, tqDebug("vgId:%d create msg to start wal scan to launch stream tasks, numOfTasks:%d, vnd restored:%d", vgId,
numOfTasks, alreadyRestored); numOfTasks, alreadyRestored);
pRunReq->head.vgId = vgId; int32_t code = streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_EXTRACT_WAL_DATA);
pRunReq->streamId = 0;
pRunReq->taskId = 0;
pRunReq->reqType = STREAM_EXEC_T_EXTRACT_WAL_DATA;
SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
tmsgPutToQueue(&pTq->pVnode->msgCb, STREAM_QUEUE, &msg);
streamMetaWUnLock(pMeta); streamMetaWUnLock(pMeta);
return 0; return code;
} }
int32_t tqStopStreamTasksAsync(STQ* pTq) { int32_t tqStopStreamTasksAsync(STQ* pTq) {
SStreamMeta* pMeta = pTq->pStreamMeta; SStreamMeta* pMeta = pTq->pStreamMeta;
int32_t vgId = pMeta->vgId; int32_t vgId = pMeta->vgId;
return streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_STOP_ALL_TASKS);
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));
if (pRunReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tqError("vgId:%d failed to create msg to stop tasks async, code:%s", vgId, terrstr());
return -1;
}
tqDebug("vgId:%d create msg to stop all tasks async", vgId);
pRunReq->head.vgId = vgId;
pRunReq->streamId = 0;
pRunReq->taskId = 0;
pRunReq->reqType = STREAM_EXEC_T_STOP_ALL_TASKS;
SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
tmsgPutToQueue(&pTq->pVnode->msgCb, STREAM_QUEUE, &msg);
return 0;
} }
int32_t setWalReaderStartOffset(SStreamTask* pTask, int32_t vgId) { int32_t setWalReaderStartOffset(SStreamTask* pTask, int32_t vgId) {
@ -408,7 +362,7 @@ int32_t doScanWalForAllTasks(SStreamMeta* pStreamMeta, bool* pScanIdle) {
if ((numOfItems > 0) || hasNewData) { if ((numOfItems > 0) || hasNewData) {
noDataInWal = false; noDataInWal = false;
code = streamSchedExec(pTask); code = streamTrySchedExec(pTask);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
streamMetaReleaseTask(pStreamMeta, pTask); streamMetaReleaseTask(pStreamMeta, pTask);
return -1; return -1;

82
source/dnode/vnode/src/tqCommon/tqCommon.c
View File

@ -17,12 +17,6 @@
#include "tq.h" #include "tq.h"
#include "tstream.h" #include "tstream.h"
typedef struct STaskUpdateEntry {
int64_t streamId;
int32_t taskId;
int32_t transId;
} STaskUpdateEntry;
typedef struct SMStreamCheckpointReadyRspMsg { typedef struct SMStreamCheckpointReadyRspMsg {
SMsgHead head; SMsgHead head;
} SMStreamCheckpointReadyRspMsg; } SMStreamCheckpointReadyRspMsg;
@ -116,22 +110,10 @@ int32_t tqStreamTaskStartAsync(SStreamMeta* pMeta, SMsgCb* cb, bool restart) {
return 0; return 0;
} }
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));
if (pRunReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tqError("vgId:%d failed to create msg to start wal scanning to launch stream tasks, code:%s", vgId, terrstr());
return -1;
}
tqDebug("vgId:%d start all %d stream task(s) async", vgId, numOfTasks); tqDebug("vgId:%d start all %d stream task(s) async", vgId, numOfTasks);
pRunReq->head.vgId = vgId;
pRunReq->streamId = 0;
pRunReq->taskId = 0;
pRunReq->reqType = restart ? STREAM_EXEC_T_RESTART_ALL_TASKS : STREAM_EXEC_T_START_ALL_TASKS;
SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)}; int32_t type = restart ? STREAM_EXEC_T_RESTART_ALL_TASKS : STREAM_EXEC_T_START_ALL_TASKS;
tmsgPutToQueue(cb, STREAM_QUEUE, &msg); return streamTaskSchedTask(cb, vgId, 0, 0, type);
return 0;
} }
int32_t tqStreamStartOneTaskAsync(SStreamMeta* pMeta, SMsgCb* cb, int64_t streamId, int32_t taskId) { int32_t tqStreamStartOneTaskAsync(SStreamMeta* pMeta, SMsgCb* cb, int64_t streamId, int32_t taskId) {
@ -143,22 +125,8 @@ int32_t tqStreamStartOneTaskAsync(SStreamMeta* pMeta, SMsgCb* cb, int64_t stream
return 0; return 0;
} }
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));
if (pRunReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tqError("vgId:%d failed to create msg to start task:0x%x, code:%s", vgId, taskId, terrstr());
return -1;
}
tqDebug("vgId:%d start task:0x%x async", vgId, taskId); tqDebug("vgId:%d start task:0x%x async", vgId, taskId);
pRunReq->head.vgId = vgId; return streamTaskSchedTask(cb, vgId, streamId, taskId, STREAM_EXEC_T_START_ONE_TASK);
pRunReq->streamId = streamId;
pRunReq->taskId = taskId;
pRunReq->reqType = STREAM_EXEC_T_START_ONE_TASK;
SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
tmsgPutToQueue(cb, STREAM_QUEUE, &msg);
return 0;
} }
int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pMsg, bool restored) { int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pMsg, bool restored) {
@ -259,6 +227,7 @@ int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pM
} }
} }
// save
if (updated) { if (updated) {
tqDebug("s-task:%s vgId:%d save task after update epset, and stop task", idstr, vgId); tqDebug("s-task:%s vgId:%d save task after update epset, and stop task", idstr, vgId);
streamMetaSaveTask(pMeta, pTask); streamMetaSaveTask(pMeta, pTask);
@ -269,22 +238,15 @@ int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pM
tqDebug("s-task:%s vgId:%d not save task since not update epset actually, stop task", idstr, vgId); tqDebug("s-task:%s vgId:%d not save task since not update epset actually, stop task", idstr, vgId);
} }
// stop
streamTaskStop(pTask); streamTaskStop(pTask);
// keep the already updated info
taosHashPut(pMeta->updateInfo.pTasks, &entry, sizeof(entry), NULL, 0);
int64_t now = taosGetTimestampMs();
if (ppHTask != NULL) { if (ppHTask != NULL) {
streamTaskStop(*ppHTask); streamTaskStop(*ppHTask);
tqDebug("s-task:%s vgId:%d task nodeEp update completed, streamTask/fill-history closed, elapsed:%" PRId64 " ms",
idstr, vgId, now - st);
taosHashPut(pMeta->updateInfo.pTasks, &(*ppHTask)->id, sizeof(pTask->id), NULL, 0);
} else {
tqDebug("s-task:%s vgId:%d, task nodeEp update completed, streamTask closed, elapsed time:%" PRId64 "ms", idstr,
vgId, now - st);
} }
// keep info
streamMetaAddIntoUpdateTaskList(pMeta, pTask, (ppHTask != NULL) ? (*ppHTask) : NULL, req.transId, st);
rsp.code = 0; rsp.code = 0;
// possibly only handle the stream task. // possibly only handle the stream task.
@ -307,10 +269,8 @@ int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pM
} else { } else {
tqDebug("vgId:%d all %d task(s) nodeEp updated and closed, transId:%d", vgId, numOfTasks, req.transId); tqDebug("vgId:%d all %d task(s) nodeEp updated and closed, transId:%d", vgId, numOfTasks, req.transId);
#if 0 #if 0
// for test purpose, to trigger the leader election taosMSleep(5000);// for test purpose, to trigger the leader election
taosMSleep(5000);
#endif #endif
tqStreamTaskStartAsync(pMeta, cb, true); tqStreamTaskStartAsync(pMeta, cb, true);
} }
} }
@ -712,26 +672,6 @@ int32_t tqStreamTaskProcessDropReq(SStreamMeta* pMeta, char* msg, int32_t msgLen
return 0; return 0;
} }
int32_t tqStreamTaskResetStatus(SStreamMeta* pMeta) {
int32_t vgId = pMeta->vgId;
int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList);
tqDebug("vgId:%d reset all %d stream task(s) status to be uninit", vgId, numOfTasks);
if (numOfTasks == 0) {
return TSDB_CODE_SUCCESS;
}
for (int32_t i = 0; i < numOfTasks; ++i) {
SStreamTaskId* pTaskId = taosArrayGet(pMeta->pTaskList, i);
STaskId id = {.streamId = pTaskId->streamId, .taskId = pTaskId->taskId};
SStreamTask** pTask = taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
streamTaskResetStatus(*pTask);
}
return 0;
}
static int32_t restartStreamTasks(SStreamMeta* pMeta, bool isLeader) { static int32_t restartStreamTasks(SStreamMeta* pMeta, bool isLeader) {
int32_t vgId = pMeta->vgId; int32_t vgId = pMeta->vgId;
int32_t code = 0; int32_t code = 0;
@ -781,7 +721,7 @@ static int32_t restartStreamTasks(SStreamMeta* pMeta, bool isLeader) {
} else { } else {
streamMetaResetStartInfo(&pMeta->startInfo); streamMetaResetStartInfo(&pMeta->startInfo);
streamMetaWUnLock(pMeta); streamMetaWUnLock(pMeta);
tqInfo("vgId:%d, follower node not start stream tasks", vgId); tqInfo("vgId:%d, follower node not start stream tasks or stream is disabled", vgId);
} }
code = terrno; code = terrno;
@ -1006,7 +946,7 @@ static int32_t tqProcessTaskResumeImpl(void* handle, SStreamTask* pTask, int64_t
} else if (level == TASK_LEVEL__SOURCE && (streamQueueGetNumOfItems(pTask->inputq.queue) == 0)) { } else if (level == TASK_LEVEL__SOURCE && (streamQueueGetNumOfItems(pTask->inputq.queue) == 0)) {
tqScanWalAsync((STQ*)handle, false); tqScanWalAsync((STQ*)handle, false);
} else { } else {
streamSchedExec(pTask); streamTrySchedExec(pTask);
} }
} else if (status == TASK_STATUS__UNINIT) { } else if (status == TASK_STATUS__UNINIT) {
// todo: fill-history task init ? // todo: fill-history task init ?

308
source/libs/stream/src/stream.c
View File

@ -1,308 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "streamInt.h"
#include "ttimer.h"
void* streamTimer = NULL;
int32_t streamTimerInit() {
streamTimer = taosTmrInit(1000, 100, 10000, "STREAM");
if (streamTimer == NULL) {
stError("init stream timer failed, code:%s", tstrerror(terrno));
return -1;
}
stInfo("init stream timer, %p", streamTimer);
return 0;
}
void streamTimerCleanUp() {
stInfo("cleanup stream timer, %p", streamTimer);
taosTmrCleanUp(streamTimer);
streamTimer = NULL;
}
tmr_h streamTimerGetInstance() {
return streamTimer;
}
char* createStreamTaskIdStr(int64_t streamId, int32_t taskId) {
char buf[128] = {0};
sprintf(buf, "0x%" PRIx64 "-0x%x", streamId, taskId);
return taosStrdup(buf);
}
static void streamSchedByTimer(void* param, void* tmrId) {
SStreamTask* pTask = (void*)param;
const char* id = pTask->id.idStr;
int32_t nextTrigger = (int32_t)pTask->info.triggerParam;
int8_t status = atomic_load_8(&pTask->schedInfo.status);
stTrace("s-task:%s in scheduler, trigger status:%d, next:%dms", id, status, nextTrigger);
if (streamTaskShouldStop(pTask) || streamTaskShouldPause(pTask)) {
stDebug("s-task:%s jump out of schedTimer", id);
return;
}
if (streamTaskGetStatus(pTask)->state == TASK_STATUS__CK) {
stDebug("s-task:%s in checkpoint procedure, not retrieve result, next:%dms", id, nextTrigger);
} else {
if (status == TASK_TRIGGER_STATUS__ACTIVE) {
SStreamTrigger* pTrigger = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0);
if (pTrigger == NULL) {
stError("s-task:%s failed to prepare retrieve data trigger, code:%s, try again in %dms", id, "out of memory",
nextTrigger);
taosTmrReset(streamSchedByTimer, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
return;
}
pTrigger->type = STREAM_INPUT__GET_RES;
pTrigger->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
if (pTrigger->pBlock == NULL) {
taosFreeQitem(pTrigger);
stError("s-task:%s failed to prepare retrieve data trigger, code:%s, try again in %dms", id, "out of memory",
nextTrigger);
taosTmrReset(streamSchedByTimer, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
return;
}
atomic_store_8(&pTask->schedInfo.status, TASK_TRIGGER_STATUS__INACTIVE);
pTrigger->pBlock->info.type = STREAM_GET_ALL;
int32_t code = streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pTrigger);
if (code != TSDB_CODE_SUCCESS) {
taosTmrReset(streamSchedByTimer, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
return;
}
streamSchedExec(pTask);
}
}
taosTmrReset(streamSchedByTimer, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
}
int32_t streamSetupScheduleTrigger(SStreamTask* pTask) {
if (pTask->info.triggerParam != 0 && pTask->info.fillHistory == 0) {
int32_t ref = atomic_add_fetch_32(&pTask->refCnt, 1);
ASSERT(ref == 2 && pTask->schedInfo.pDelayTimer == NULL);
stDebug("s-task:%s setup scheduler trigger, delay:%" PRId64 " ms", pTask->id.idStr, pTask->info.triggerParam);
pTask->schedInfo.pDelayTimer = taosTmrStart(streamSchedByTimer, (int32_t)pTask->info.triggerParam, pTask, streamTimer);
pTask->schedInfo.status = TASK_TRIGGER_STATUS__INACTIVE;
}
return 0;
}
int32_t streamSchedExec(SStreamTask* pTask) {
if (streamTaskSetSchedStatusWait(pTask)) {
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));
if (pRunReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
/*int8_t status = */streamTaskSetSchedStatusInactive(pTask);
stError("failed to create msg to aunch s-task:%s, reason out of memory", pTask->id.idStr);
return -1;
}
pRunReq->head.vgId = pTask->info.nodeId;
pRunReq->streamId = pTask->id.streamId;
pRunReq->taskId = pTask->id.taskId;
stDebug("trigger to run s-task:%s", pTask->id.idStr);
SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
tmsgPutToQueue(pTask->pMsgCb, STREAM_QUEUE, &msg);
} else {
stTrace("s-task:%s not launch task since sched status:%d", pTask->id.idStr, pTask->status.schedStatus);
}
return 0;
}
static int32_t buildDispatchRsp(const SStreamTask* pTask, const SStreamDispatchReq* pReq, int32_t status, void** pBuf) {
*pBuf = rpcMallocCont(sizeof(SMsgHead) + sizeof(SStreamDispatchRsp));
if (*pBuf == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
((SMsgHead*)(*pBuf))->vgId = htonl(pReq->upstreamNodeId);
ASSERT(((SMsgHead*)(*pBuf))->vgId != 0);
SStreamDispatchRsp* pDispatchRsp = POINTER_SHIFT((*pBuf), sizeof(SMsgHead));
pDispatchRsp->stage = htobe64(pReq->stage);
pDispatchRsp->msgId = htonl(pReq->msgId);
pDispatchRsp->inputStatus = status;
pDispatchRsp->streamId = htobe64(pReq->streamId);
pDispatchRsp->upstreamNodeId = htonl(pReq->upstreamNodeId);
pDispatchRsp->upstreamTaskId = htonl(pReq->upstreamTaskId);
pDispatchRsp->downstreamNodeId = htonl(pTask->info.nodeId);
pDispatchRsp->downstreamTaskId = htonl(pTask->id.taskId);
return TSDB_CODE_SUCCESS;
}
static int32_t streamTaskAppendInputBlocks(SStreamTask* pTask, const SStreamDispatchReq* pReq) {
int8_t status = 0;
SStreamDataBlock* pBlock = createStreamBlockFromDispatchMsg(pReq, pReq->type, pReq->srcVgId);
if (pBlock == NULL) {
streamTaskInputFail(pTask);
status = TASK_INPUT_STATUS__FAILED;
stError("vgId:%d, s-task:%s failed to receive dispatch msg, reason: out of memory", pTask->pMeta->vgId,
pTask->id.idStr);
} else {
if (pBlock->type == STREAM_INPUT__TRANS_STATE) {
pTask->status.appendTranstateBlock = true;
}
int32_t code = streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pBlock);
// input queue is full, upstream is blocked now
status = (code == TSDB_CODE_SUCCESS) ? TASK_INPUT_STATUS__NORMAL : TASK_INPUT_STATUS__BLOCKED;
}
return status;
}
int32_t streamTaskEnqueueRetrieve(SStreamTask* pTask, SStreamRetrieveReq* pReq) {
SStreamDataBlock* pData = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SStreamDataBlock));
int8_t status = TASK_INPUT_STATUS__NORMAL;
// enqueue
if (pData != NULL) {
stDebug("s-task:%s (child %d) recv retrieve req from task:0x%x(vgId:%d), reqId:0x%" PRIx64, pTask->id.idStr,
pTask->info.selfChildId, pReq->srcTaskId, pReq->srcNodeId, pReq->reqId);
pData->type = STREAM_INPUT__DATA_RETRIEVE;
pData->srcVgId = 0;
streamRetrieveReqToData(pReq, pData);
if (streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pData) == 0) {
status = TASK_INPUT_STATUS__NORMAL;
} else {
status = TASK_INPUT_STATUS__FAILED;
}
} else { // todo handle oom
/*streamTaskInputFail(pTask);*/
/*status = TASK_INPUT_STATUS__FAILED;*/
}
return status == TASK_INPUT_STATUS__NORMAL ? 0 : -1;
}
int32_t streamProcessDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pReq, SRpcMsg* pRsp) {
int32_t status = 0;
SStreamMeta* pMeta = pTask->pMeta;
const char* id = pTask->id.idStr;
stDebug("s-task:%s receive dispatch msg from taskId:0x%x(vgId:%d), msgLen:%" PRId64 ", msgId:%d", id,
pReq->upstreamTaskId, pReq->upstreamNodeId, pReq->totalLen, pReq->msgId);
SStreamChildEpInfo* pInfo = streamTaskGetUpstreamTaskEpInfo(pTask, pReq->upstreamTaskId);
ASSERT(pInfo != NULL);
if (pMeta->role == NODE_ROLE_FOLLOWER) {
stError("s-task:%s task on follower received dispatch msgs, dispatch msg rejected", id);
status = TASK_INPUT_STATUS__REFUSED;
} else {
if (pReq->stage > pInfo->stage) {
// upstream task has restarted/leader-follower switch/transferred to other dnodes
stError("s-task:%s upstream task:0x%x (vgId:%d) has restart/leader-switch/vnode-transfer, prev stage:%" PRId64
", current:%" PRId64 " dispatch msg rejected",
id, pReq->upstreamTaskId, pReq->upstreamNodeId, pInfo->stage, pReq->stage);
status = TASK_INPUT_STATUS__REFUSED;
} else {
if (!pInfo->dataAllowed) {
stWarn("s-task:%s data from task:0x%x is denied, since inputQ is closed for it", id, pReq->upstreamTaskId);
status = TASK_INPUT_STATUS__BLOCKED;
} else {
// This task has received the checkpoint req from the upstream task, from which all the messages should be
// blocked. Note that there is no race condition here.
if (pReq->type == STREAM_INPUT__CHECKPOINT_TRIGGER) {
atomic_add_fetch_32(&pTask->upstreamInfo.numOfClosed, 1);
streamTaskCloseUpstreamInput(pTask, pReq->upstreamTaskId);
stDebug("s-task:%s close inputQ for upstream:0x%x, msgId:%d", id, pReq->upstreamTaskId, pReq->msgId);
} else if (pReq->type == STREAM_INPUT__TRANS_STATE) {
atomic_add_fetch_32(&pTask->upstreamInfo.numOfClosed, 1);
streamTaskCloseUpstreamInput(pTask, pReq->upstreamTaskId);
// disable the related stream task here to avoid it to receive the newly arrived data after the transfer-state
STaskId* pRelTaskId = &pTask->streamTaskId;
SStreamTask* pStreamTask = streamMetaAcquireTask(pMeta, pRelTaskId->streamId, pRelTaskId->taskId);
if (pStreamTask != NULL) {
atomic_add_fetch_32(&pStreamTask->upstreamInfo.numOfClosed, 1);
streamTaskCloseUpstreamInput(pStreamTask, pReq->upstreamRelTaskId);
streamMetaReleaseTask(pMeta, pStreamTask);
}
stDebug("s-task:%s close inputQ for upstream:0x%x since trans-state msgId:%d recv, rel stream-task:0x%" PRIx64
" close inputQ for upstream:0x%x",
id, pReq->upstreamTaskId, pReq->msgId, pTask->streamTaskId.taskId, pReq->upstreamRelTaskId);
}
status = streamTaskAppendInputBlocks(pTask, pReq);
}
}
}
// disable the data from upstream tasks
// if (streamTaskGetStatus(pTask)->state == TASK_STATUS__HALT) {
// status = TASK_INPUT_STATUS__BLOCKED;
// }
{
// do send response with the input status
int32_t code = buildDispatchRsp(pTask, pReq, status, &pRsp->pCont);
if (code != TSDB_CODE_SUCCESS) {
stError("s-task:%s failed to build dispatch rsp, msgId:%d, code:%s", id, pReq->msgId, tstrerror(code));
terrno = code;
return code;
}
pRsp->contLen = sizeof(SMsgHead) + sizeof(SStreamDispatchRsp);
tmsgSendRsp(pRsp);
}
streamSchedExec(pTask);
return 0;
}
int32_t streamProcessRetrieveReq(SStreamTask* pTask, SStreamRetrieveReq* pReq) {
int32_t code = streamTaskEnqueueRetrieve(pTask, pReq);
if(code != 0){
return code;
}
return streamSchedExec(pTask);
}
void streamTaskInputFail(SStreamTask* pTask) { atomic_store_8(&pTask->inputq.status, TASK_INPUT_STATUS__FAILED); }
SStreamChildEpInfo* streamTaskGetUpstreamTaskEpInfo(SStreamTask* pTask, int32_t taskId) {
int32_t num = taosArrayGetSize(pTask->upstreamInfo.pList);
for (int32_t i = 0; i < num; ++i) {
SStreamChildEpInfo* pInfo = taosArrayGetP(pTask->upstreamInfo.pList, i);
if (pInfo->taskId == taskId) {
return pInfo;
}
}
stError("s-task:%s failed to find upstream task:0x%x", pTask->id.idStr, taskId);
return NULL;
}

50
source/libs/stream/src/streamCheckStatus.c
View File

@ -341,24 +341,24 @@ int32_t streamTaskCompleteCheckRsp(STaskCheckInfo* pInfo, bool lock, const char*
taosThreadMutexLock(&pInfo->checkInfoLock); taosThreadMutexLock(&pInfo->checkInfoLock);
} }
if (!pInfo->inCheckProcess) { if (pInfo->inCheckProcess) {
// stWarn("s-task:%s already not in-check-procedure", id); int64_t el = (pInfo->startTs != 0) ? (taosGetTimestampMs() - pInfo->startTs) : 0;
stDebug("s-task:%s clear the in check-rsp flag, set the check-rsp done, elapsed time:%" PRId64 " ms", id, el);
pInfo->startTs = 0;
pInfo->timeoutStartTs = 0;
pInfo->notReadyTasks = 0;
pInfo->inCheckProcess = 0;
pInfo->stopCheckProcess = 0;
pInfo->notReadyRetryCount = 0;
pInfo->timeoutRetryCount = 0;
taosArrayClear(pInfo->pList);
} else {
stDebug("s-task:%s already not in check-rsp procedure", id);
} }
int64_t el = (pInfo->startTs != 0) ? (taosGetTimestampMs() - pInfo->startTs) : 0;
stDebug("s-task:%s clear the in check-rsp flag, not in check-rsp anymore, elapsed time:%" PRId64 " ms", id, el);
pInfo->startTs = 0;
pInfo->timeoutStartTs = 0;
pInfo->notReadyTasks = 0;
pInfo->inCheckProcess = 0;
pInfo->stopCheckProcess = 0;
pInfo->notReadyRetryCount = 0;
pInfo->timeoutRetryCount = 0;
taosArrayClear(pInfo->pList);
if (lock) { if (lock) {
taosThreadMutexUnlock(&pInfo->checkInfoLock); taosThreadMutexUnlock(&pInfo->checkInfoLock);
} }
@ -527,23 +527,7 @@ void handleNotReadyDownstreamTask(SStreamTask* pTask, SArray* pNotReadyList) {
// The restart of all tasks requires that all tasks should not have active timer for now. Therefore, the execution // The restart of all tasks requires that all tasks should not have active timer for now. Therefore, the execution
// of restart in timer thread will result in a dead lock. // of restart in timer thread will result in a dead lock.
int32_t addDownstreamFailedStatusResultAsync(SMsgCb* pMsgCb, int32_t vgId, int64_t streamId, int32_t taskId) { int32_t addDownstreamFailedStatusResultAsync(SMsgCb* pMsgCb, int32_t vgId, int64_t streamId, int32_t taskId) {
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq)); return streamTaskSchedTask(pMsgCb, vgId, streamId, taskId, STREAM_EXEC_T_ADD_FAILED_TASK);
if (pRunReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
stError("vgId:%d failed to create msg to stop tasks async, code:%s", vgId, terrstr());
return -1;
}
stDebug("vgId:%d create msg add failed s-task:0x%x", vgId, taskId);
pRunReq->head.vgId = vgId;
pRunReq->streamId = streamId;
pRunReq->taskId = taskId;
pRunReq->reqType = STREAM_EXEC_T_ADD_FAILED_TASK;
SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
tmsgPutToQueue(pMsgCb, STREAM_QUEUE, &msg);
return 0;
} }
// this function is executed in timer thread // this function is executed in timer thread

12
source/libs/stream/src/streamCheckpoint.c
View File

@ -30,8 +30,8 @@ typedef struct {
static int32_t downloadCheckpointDataByName(const char* id, const char* fname, const char* dstName); static int32_t downloadCheckpointDataByName(const char* id, const char* fname, const char* dstName);
static int32_t deleteCheckpointFile(const char* id, const char* name); static int32_t deleteCheckpointFile(const char* id, const char* name);
static int32_t streamTaskBackupCheckpoint(char* id, char* path); static int32_t streamTaskBackupCheckpoint(const char* id, const char* path);
static int32_t deleteCheckpoint(char* id); static int32_t deleteCheckpoint(const char* id);
int32_t tEncodeStreamCheckpointSourceReq(SEncoder* pEncoder, const SStreamCheckpointSourceReq* pReq) { int32_t tEncodeStreamCheckpointSourceReq(SEncoder* pEncoder, const SStreamCheckpointSourceReq* pReq) {
if (tStartEncode(pEncoder) < 0) return -1; if (tStartEncode(pEncoder) < 0) return -1;
@ -157,7 +157,7 @@ static int32_t appendCheckpointIntoInputQ(SStreamTask* pTask, int32_t checkpoint
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
streamSchedExec(pTask); streamTrySchedExec(pTask);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -578,7 +578,7 @@ int32_t streamTaskBuildCheckpoint(SStreamTask* pTask) {
return code; return code;
} }
static int32_t uploadCheckpointToS3(char* id, char* path) { static int32_t uploadCheckpointToS3(const char* id, const char* path) {
TdDirPtr pDir = taosOpenDir(path); TdDirPtr pDir = taosOpenDir(path);
if (pDir == NULL) return -1; if (pDir == NULL) return -1;
@ -631,7 +631,7 @@ ECHECKPOINT_BACKUP_TYPE streamGetCheckpointBackupType() {
} }
} }
int32_t streamTaskBackupCheckpoint(char* id, char* path) { int32_t streamTaskBackupCheckpoint(const char* id, const char* path) {
if (id == NULL || path == NULL || strlen(id) == 0 || strlen(path) == 0 || strlen(path) >= PATH_MAX) { if (id == NULL || path == NULL || strlen(id) == 0 || strlen(path) == 0 || strlen(path) >= PATH_MAX) {
stError("streamTaskBackupCheckpoint parameters invalid"); stError("streamTaskBackupCheckpoint parameters invalid");
return -1; return -1;
@ -675,7 +675,7 @@ int32_t streamTaskDownloadCheckpointData(char* id, char* path) {
return 0; return 0;
} }
int32_t deleteCheckpoint(char* id) { int32_t deleteCheckpoint(const char* id) {
if (id == NULL || strlen(id) == 0) { if (id == NULL || strlen(id) == 0) {
stError("deleteCheckpoint parameters invalid"); stError("deleteCheckpoint parameters invalid");
return -1; return -1;

123
source/libs/stream/src/streamDispatch.c
View File

@ -1137,6 +1137,129 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, i
return 0; return 0;
} }
static int32_t buildDispatchRsp(const SStreamTask* pTask, const SStreamDispatchReq* pReq, int32_t status, void** pBuf) {
*pBuf = rpcMallocCont(sizeof(SMsgHead) + sizeof(SStreamDispatchRsp));
if (*pBuf == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
((SMsgHead*)(*pBuf))->vgId = htonl(pReq->upstreamNodeId);
ASSERT(((SMsgHead*)(*pBuf))->vgId != 0);
SStreamDispatchRsp* pDispatchRsp = POINTER_SHIFT((*pBuf), sizeof(SMsgHead));
pDispatchRsp->stage = htobe64(pReq->stage);
pDispatchRsp->msgId = htonl(pReq->msgId);
pDispatchRsp->inputStatus = status;
pDispatchRsp->streamId = htobe64(pReq->streamId);
pDispatchRsp->upstreamNodeId = htonl(pReq->upstreamNodeId);
pDispatchRsp->upstreamTaskId = htonl(pReq->upstreamTaskId);
pDispatchRsp->downstreamNodeId = htonl(pTask->info.nodeId);
pDispatchRsp->downstreamTaskId = htonl(pTask->id.taskId);
return TSDB_CODE_SUCCESS;
}
static int32_t streamTaskAppendInputBlocks(SStreamTask* pTask, const SStreamDispatchReq* pReq) {
int8_t status = 0;
SStreamDataBlock* pBlock = createStreamBlockFromDispatchMsg(pReq, pReq->type, pReq->srcVgId);
if (pBlock == NULL) {
streamTaskInputFail(pTask);
status = TASK_INPUT_STATUS__FAILED;
stError("vgId:%d, s-task:%s failed to receive dispatch msg, reason: out of memory", pTask->pMeta->vgId,
pTask->id.idStr);
} else {
if (pBlock->type == STREAM_INPUT__TRANS_STATE) {
pTask->status.appendTranstateBlock = true;
}
int32_t code = streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pBlock);
// input queue is full, upstream is blocked now
status = (code == TSDB_CODE_SUCCESS) ? TASK_INPUT_STATUS__NORMAL : TASK_INPUT_STATUS__BLOCKED;
}
return status;
}
int32_t streamProcessDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pReq, SRpcMsg* pRsp) {
int32_t status = 0;
SStreamMeta* pMeta = pTask->pMeta;
const char* id = pTask->id.idStr;
stDebug("s-task:%s receive dispatch msg from taskId:0x%x(vgId:%d), msgLen:%" PRId64 ", msgId:%d", id,
pReq->upstreamTaskId, pReq->upstreamNodeId, pReq->totalLen, pReq->msgId);
SStreamChildEpInfo* pInfo = streamTaskGetUpstreamTaskEpInfo(pTask, pReq->upstreamTaskId);
ASSERT(pInfo != NULL);
if (pMeta->role == NODE_ROLE_FOLLOWER) {
stError("s-task:%s task on follower received dispatch msgs, dispatch msg rejected", id);
status = TASK_INPUT_STATUS__REFUSED;
} else {
if (pReq->stage > pInfo->stage) {
// upstream task has restarted/leader-follower switch/transferred to other dnodes
stError("s-task:%s upstream task:0x%x (vgId:%d) has restart/leader-switch/vnode-transfer, prev stage:%" PRId64
", current:%" PRId64 " dispatch msg rejected",
id, pReq->upstreamTaskId, pReq->upstreamNodeId, pInfo->stage, pReq->stage);
status = TASK_INPUT_STATUS__REFUSED;
} else {
if (!pInfo->dataAllowed) {
stWarn("s-task:%s data from task:0x%x is denied, since inputQ is closed for it", id, pReq->upstreamTaskId);
status = TASK_INPUT_STATUS__BLOCKED;
} else {
// This task has received the checkpoint req from the upstream task, from which all the messages should be
// blocked. Note that there is no race condition here.
if (pReq->type == STREAM_INPUT__CHECKPOINT_TRIGGER) {
atomic_add_fetch_32(&pTask->upstreamInfo.numOfClosed, 1);
streamTaskCloseUpstreamInput(pTask, pReq->upstreamTaskId);
stDebug("s-task:%s close inputQ for upstream:0x%x, msgId:%d", id, pReq->upstreamTaskId, pReq->msgId);
} else if (pReq->type == STREAM_INPUT__TRANS_STATE) {
atomic_add_fetch_32(&pTask->upstreamInfo.numOfClosed, 1);
streamTaskCloseUpstreamInput(pTask, pReq->upstreamTaskId);
// disable the related stream task here to avoid it to receive the newly arrived data after the transfer-state
STaskId* pRelTaskId = &pTask->streamTaskId;
SStreamTask* pStreamTask = streamMetaAcquireTask(pMeta, pRelTaskId->streamId, pRelTaskId->taskId);
if (pStreamTask != NULL) {
atomic_add_fetch_32(&pStreamTask->upstreamInfo.numOfClosed, 1);
streamTaskCloseUpstreamInput(pStreamTask, pReq->upstreamRelTaskId);
streamMetaReleaseTask(pMeta, pStreamTask);
}
stDebug("s-task:%s close inputQ for upstream:0x%x since trans-state msgId:%d recv, rel stream-task:0x%" PRIx64
" close inputQ for upstream:0x%x",
id, pReq->upstreamTaskId, pReq->msgId, pTask->streamTaskId.taskId, pReq->upstreamRelTaskId);
}
status = streamTaskAppendInputBlocks(pTask, pReq);
}
}
}
// disable the data from upstream tasks
// if (streamTaskGetStatus(pTask)->state == TASK_STATUS__HALT) {
// status = TASK_INPUT_STATUS__BLOCKED;
// }
{
// do send response with the input status
int32_t code = buildDispatchRsp(pTask, pReq, status, &pRsp->pCont);
if (code != TSDB_CODE_SUCCESS) {
stError("s-task:%s failed to build dispatch rsp, msgId:%d, code:%s", id, pReq->msgId, tstrerror(code));
terrno = code;
return code;
}
pRsp->contLen = sizeof(SMsgHead) + sizeof(SStreamDispatchRsp);
tmsgSendRsp(pRsp);
}
streamTrySchedExec(pTask);
return 0;
}
int32_t tEncodeStreamTaskUpdateMsg(SEncoder* pEncoder, const SStreamTaskNodeUpdateMsg* pMsg) { int32_t tEncodeStreamTaskUpdateMsg(SEncoder* pEncoder, const SStreamTaskNodeUpdateMsg* pMsg) {
if (tStartEncode(pEncoder) < 0) return -1; if (tStartEncode(pEncoder) < 0) return -1;
if (tEncodeI64(pEncoder, pMsg->streamId) < 0) return -1; if (tEncodeI64(pEncoder, pMsg->streamId) < 0) return -1;

75
source/libs/stream/src/streamExec.c
View File

@ -533,8 +533,7 @@ int32_t streamProcessTransstateBlock(SStreamTask* pTask, SStreamDataBlock* pBloc
return code; return code;
} }
static void setTaskSchedInfo(SStreamTask* pTask, int32_t idleTime) { pTask->status.schedIdleTime = idleTime; } //static void streamTaskSetIdleInfo(SStreamTask* pTask, int32_t idleTime) { pTask->status.schedIdleTime = idleTime; }
static void clearTaskSchedInfo(SStreamTask* pTask) { pTask->status.schedIdleTime = 0; }
static void setLastExecTs(SStreamTask* pTask, int64_t ts) { pTask->status.lastExecTs = ts; } static void setLastExecTs(SStreamTask* pTask, int64_t ts) { pTask->status.lastExecTs = ts; }
/** /**
@ -559,26 +558,26 @@ static int32_t doStreamExecTask(SStreamTask* pTask) {
if (streamQueueIsFull(pTask->outputq.queue)) { if (streamQueueIsFull(pTask->outputq.queue)) {
stWarn("s-task:%s outputQ is full, idle for 500ms and retry", id); stWarn("s-task:%s outputQ is full, idle for 500ms and retry", id);
setTaskSchedInfo(pTask, 500); streamTaskSetIdleInfo(pTask, 500);
return 0; return 0;
} }
if (pTask->inputq.status == TASK_INPUT_STATUS__BLOCKED) { if (pTask->inputq.status == TASK_INPUT_STATUS__BLOCKED) {
stWarn("s-task:%s downstream task inputQ blocked, idle for 1sec and retry", id); stWarn("s-task:%s downstream task inputQ blocked, idle for 1sec and retry", id);
setTaskSchedInfo(pTask, 1000); streamTaskSetIdleInfo(pTask, 1000);
return 0; return 0;
} }
if (taosGetTimestampMs() - pTask->status.lastExecTs < MIN_INVOKE_INTERVAL) { if (taosGetTimestampMs() - pTask->status.lastExecTs < MIN_INVOKE_INTERVAL) {
stDebug("s-task:%s invoke with high frequency, idle and retry exec in 50ms", id); stDebug("s-task:%s invoke with high frequency, idle and retry exec in 50ms", id);
setTaskSchedInfo(pTask, MIN_INVOKE_INTERVAL); streamTaskSetIdleInfo(pTask, MIN_INVOKE_INTERVAL);
return 0; return 0;
} }
EExtractDataCode ret = streamTaskGetDataFromInputQ(pTask, &pInput, &numOfBlocks, &blockSize); EExtractDataCode ret = streamTaskGetDataFromInputQ(pTask, &pInput, &numOfBlocks, &blockSize);
if (ret == EXEC_AFTER_IDLE) { if (ret == EXEC_AFTER_IDLE) {
ASSERT(pInput == NULL && numOfBlocks == 0); ASSERT(pInput == NULL && numOfBlocks == 0);
setTaskSchedInfo(pTask, MIN_INVOKE_INTERVAL); streamTaskSetIdleInfo(pTask, MIN_INVOKE_INTERVAL);
return 0; return 0;
} else { } else {
if (pInput == NULL) { if (pInput == NULL) {
@ -720,66 +719,6 @@ bool streamTaskReadyToRun(const SStreamTask* pTask, char** pStatus) {
} }
} }
static void doStreamExecTaskHelper(void* param, void* tmrId) {
SStreamTask* pTask = (SStreamTask*)param;
SStreamTaskState* p = streamTaskGetStatus(pTask);
if (p->state == TASK_STATUS__DROPPING || p->state == TASK_STATUS__STOP) {
streamTaskSetSchedStatusInactive(pTask);
int32_t ref = atomic_sub_fetch_32(&pTask->status.timerActive, 1);
stDebug("s-task:%s status:%s not resume task, ref:%d", pTask->id.idStr, p->name, ref);
streamMetaReleaseTask(pTask->pMeta, pTask);
return;
}
// task resume running
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));
if (pRunReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
/*int8_t status = */streamTaskSetSchedStatusInactive(pTask);
int32_t ref = atomic_sub_fetch_32(&pTask->status.timerActive, 1);
stError("failed to create msg to resume s-task:%s, reason out of memory, ref:%d", pTask->id.idStr, ref);
streamMetaReleaseTask(pTask->pMeta, pTask);
return;
}
pRunReq->head.vgId = pTask->info.nodeId;
pRunReq->streamId = pTask->id.streamId;
pRunReq->taskId = pTask->id.taskId;
pRunReq->reqType = STREAM_EXEC_T_RESUME_TASK;
int32_t ref = atomic_sub_fetch_32(&pTask->status.timerActive, 1);
stDebug("trigger to resume s-task:%s after being idled for %dms, ref:%d", pTask->id.idStr, pTask->status.schedIdleTime, ref);
SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
tmsgPutToQueue(pTask->pMsgCb, STREAM_QUEUE, &msg);
// release the task ref count
clearTaskSchedInfo(pTask);
streamMetaReleaseTask(pTask->pMeta, pTask);
}
static int32_t schedTaskInFuture(SStreamTask* pTask) {
int32_t ref = atomic_add_fetch_32(&pTask->status.timerActive, 1);
stDebug("s-task:%s task should idle, add into timer to retry in %dms, ref:%d", pTask->id.idStr,
pTask->status.schedIdleTime, ref);
// add one ref count for task
/*SStreamTask* pAddRefTask = */streamMetaAcquireOneTask(pTask);
if (pTask->schedInfo.pIdleTimer == NULL) {
pTask->schedInfo.pIdleTimer = taosTmrStart(doStreamExecTaskHelper, pTask->status.schedIdleTime, pTask, streamTimer);
} else {
taosTmrReset(doStreamExecTaskHelper, pTask->status.schedIdleTime, pTask, streamTimer, &pTask->schedInfo.pIdleTimer);
}
return TSDB_CODE_SUCCESS;
}
int32_t streamResumeTask(SStreamTask* pTask) { int32_t streamResumeTask(SStreamTask* pTask) {
ASSERT(pTask->status.schedStatus == TASK_SCHED_STATUS__ACTIVE); ASSERT(pTask->status.schedStatus == TASK_SCHED_STATUS__ACTIVE);
const char* id = pTask->id.idStr; const char* id = pTask->id.idStr;
@ -793,7 +732,7 @@ int32_t streamResumeTask(SStreamTask* pTask) {
int32_t numOfItems = streamQueueGetNumOfItems(pTask->inputq.queue); int32_t numOfItems = streamQueueGetNumOfItems(pTask->inputq.queue);
if ((numOfItems == 0) || streamTaskShouldStop(pTask) || streamTaskShouldPause(pTask)) { if ((numOfItems == 0) || streamTaskShouldStop(pTask) || streamTaskShouldPause(pTask)) {
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE); atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
clearTaskSchedInfo(pTask); streamTaskClearSchedIdleInfo(pTask);
taosThreadMutexUnlock(&pTask->lock); taosThreadMutexUnlock(&pTask->lock);
setLastExecTs(pTask, taosGetTimestampMs()); setLastExecTs(pTask, taosGetTimestampMs());
@ -806,7 +745,7 @@ int32_t streamResumeTask(SStreamTask* pTask) {
} else { } else {
// check if this task needs to be idle for a while // check if this task needs to be idle for a while
if (pTask->status.schedIdleTime > 0) { if (pTask->status.schedIdleTime > 0) {
schedTaskInFuture(pTask); streamTaskResumeInFuture(pTask);
taosThreadMutexUnlock(&pTask->lock); taosThreadMutexUnlock(&pTask->lock);
setLastExecTs(pTask, taosGetTimestampMs()); setLastExecTs(pTask, taosGetTimestampMs());

57
source/libs/stream/src/streamMeta.c
View File

@ -373,6 +373,7 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
pMeta->numOfPausedTasks = 0; pMeta->numOfPausedTasks = 0;
pMeta->numOfStreamTasks = 0; pMeta->numOfStreamTasks = 0;
pMeta->closeFlag = false;
stInfo("vgId:%d open stream meta succ, latest checkpoint:%" PRId64 ", stage:%" PRId64, vgId, pMeta->chkpId, stage); stInfo("vgId:%d open stream meta succ, latest checkpoint:%" PRId64 ", stage:%" PRId64, vgId, pMeta->chkpId, stage);
@ -1215,7 +1216,7 @@ void metaHbToMnode(void* param, void* tmrId) {
} }
// need to stop, stop now // need to stop, stop now
if (pMeta->pHbInfo->stopFlag == STREAM_META_WILL_STOP) { if (pMeta->pHbInfo->stopFlag == STREAM_META_WILL_STOP) { // todo refactor: not need this now, use closeFlag in Meta
pMeta->pHbInfo->stopFlag = STREAM_META_OK_TO_STOP; pMeta->pHbInfo->stopFlag = STREAM_META_OK_TO_STOP;
stDebug("vgId:%d jump out of meta timer", pMeta->vgId); stDebug("vgId:%d jump out of meta timer", pMeta->vgId);
taosReleaseRef(streamMetaId, rid); taosReleaseRef(streamMetaId, rid);
@ -1281,6 +1282,8 @@ void streamMetaNotifyClose(SStreamMeta* pMeta) {
streamMetaWLock(pMeta); streamMetaWLock(pMeta);
pMeta->closeFlag = true;
void* pIter = NULL; void* pIter = NULL;
while (1) { while (1) {
pIter = taosHashIterate(pMeta->pTasksMap, pIter); pIter = taosHashIterate(pMeta->pTasksMap, pIter);
@ -1304,7 +1307,7 @@ void streamMetaNotifyClose(SStreamMeta* pMeta) {
} }
} }
stDebug("vgId:%d start to check all tasks", vgId); stDebug("vgId:%d start to check all tasks for closing", vgId);
int64_t st = taosGetTimestampMs(); int64_t st = taosGetTimestampMs();
while (streamMetaTaskInTimer(pMeta)) { while (streamMetaTaskInTimer(pMeta)) {
@ -1438,35 +1441,47 @@ void streamMetaUpdateStageRole(SStreamMeta* pMeta, int64_t stage, bool isLeader)
} }
} }
static SArray* prepareBeforeStartTasks(SStreamMeta* pMeta) { static int32_t prepareBeforeStartTasks(SStreamMeta* pMeta, SArray** pList, int64_t now) {
streamMetaWLock(pMeta); streamMetaWLock(pMeta);
SArray* pTaskList = taosArrayDup(pMeta->pTaskList, NULL); if (pMeta->closeFlag) {
streamMetaWUnLock(pMeta);
stError("vgId:%d vnode is closed, not start check task(s) downstream status", pMeta->vgId);
return -1;
}
*pList = taosArrayDup(pMeta->pTaskList, NULL);
taosHashClear(pMeta->startInfo.pReadyTaskSet); taosHashClear(pMeta->startInfo.pReadyTaskSet);
taosHashClear(pMeta->startInfo.pFailedTaskSet); taosHashClear(pMeta->startInfo.pFailedTaskSet);
pMeta->startInfo.startTs = taosGetTimestampMs(); pMeta->startInfo.startTs = now;
streamMetaResetTaskStatus(pMeta); streamMetaResetTaskStatus(pMeta);
streamMetaWUnLock(pMeta); streamMetaWUnLock(pMeta);
return pTaskList; return TSDB_CODE_SUCCESS;
} }
int32_t streamMetaStartAllTasks(SStreamMeta* pMeta) { int32_t streamMetaStartAllTasks(SStreamMeta* pMeta) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
int32_t vgId = pMeta->vgId; int32_t vgId = pMeta->vgId;
int64_t now = taosGetTimestampMs();
int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList); int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList);
stInfo("vgId:%d start to check all %d stream task(s) downstream status", vgId, numOfTasks); stInfo("vgId:%d start to check all %d stream task(s) downstream status, start ts:%"PRId64, vgId, numOfTasks, now);
if (numOfTasks == 0) { if (numOfTasks == 0) {
stInfo("vgId:%d start tasks completed", pMeta->vgId); stInfo("vgId:%d no tasks to be started", pMeta->vgId);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int64_t now = taosGetTimestampMs(); SArray* pTaskList = NULL;
code = prepareBeforeStartTasks(pMeta, &pTaskList, now);
if (code != TSDB_CODE_SUCCESS) {
ASSERT(pTaskList == NULL);
return TSDB_CODE_SUCCESS;
}
SArray* pTaskList = prepareBeforeStartTasks(pMeta);
numOfTasks = taosArrayGetSize(pTaskList); numOfTasks = taosArrayGetSize(pTaskList);
// broadcast the check downstream tasks msg // broadcast the check downstream tasks msg
@ -1742,4 +1757,26 @@ int32_t streamMetaAddFailedTask(SStreamMeta* pMeta, int64_t streamId, int32_t ta
} }
return code; return code;
}
void streamMetaAddIntoUpdateTaskList(SStreamMeta* pMeta, SStreamTask* pTask, SStreamTask* pHTask, int32_t transId,
int64_t startTs) {
const char* id = pTask->id.idStr;
int32_t vgId = pTask->pMeta->vgId;
// keep the already updated info
STaskUpdateEntry entry = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId, .transId = transId};
taosHashPut(pMeta->updateInfo.pTasks, &entry, sizeof(entry), NULL, 0);
int64_t el = taosGetTimestampMs() - startTs;
if (pHTask != NULL) {
STaskUpdateEntry hEntry = {.streamId = pHTask->id.streamId, .taskId = pHTask->id.taskId, .transId = transId};
taosHashPut(pMeta->updateInfo.pTasks, &hEntry, sizeof(hEntry), NULL, 0);
stDebug("s-task:%s vgId:%d transId:%d task nodeEp update completed, streamTask/hTask closed, elapsed:%" PRId64
" ms", id, vgId, transId, el);
} else {
stDebug("s-task:%s vgId:%d transId:%d task nodeEp update completed, streamTask closed, elapsed time:%" PRId64 "ms",
id, vgId, transId, el);
}
} }

4
source/libs/stream/src/streamQueue.c
View File

@ -424,4 +424,6 @@ void streamTaskPutbackToken(STokenBucket* pBucket) {
// size in KB // size in KB
void streamTaskConsumeQuota(STokenBucket* pBucket, int32_t bytes) { void streamTaskConsumeQuota(STokenBucket* pBucket, int32_t bytes) {
pBucket->quotaRemain -= SIZE_IN_MiB(bytes); pBucket->quotaRemain -= SIZE_IN_MiB(bytes);
} }
void streamTaskInputFail(SStreamTask* pTask) { atomic_store_8(&pTask->inputq.status, TASK_INPUT_STATUS__FAILED); }

165
source/libs/stream/src/streamSched.c Normal file
View File

@ -0,0 +1,165 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "streamInt.h"
#include "ttimer.h"
static void streamTaskResumeHelper(void* param, void* tmrId);
static void streamSchedByTimer(void* param, void* tmrId);
int32_t streamSetupScheduleTrigger(SStreamTask* pTask) {
if (pTask->info.triggerParam != 0 && pTask->info.fillHistory == 0) {
int32_t ref = atomic_add_fetch_32(&pTask->refCnt, 1);
ASSERT(ref == 2 && pTask->schedInfo.pDelayTimer == NULL);
stDebug("s-task:%s setup scheduler trigger, delay:%" PRId64 " ms", pTask->id.idStr, pTask->info.triggerParam);
pTask->schedInfo.pDelayTimer = taosTmrStart(streamSchedByTimer, (int32_t)pTask->info.triggerParam, pTask, streamTimer);
pTask->schedInfo.status = TASK_TRIGGER_STATUS__INACTIVE;
}
return 0;
}
int32_t streamTrySchedExec(SStreamTask* pTask) {
if (streamTaskSetSchedStatusWait(pTask)) {
streamTaskSchedTask(pTask->pMsgCb, pTask->info.nodeId, pTask->id.streamId, pTask->id.taskId, 0);
} else {
stTrace("s-task:%s not launch task since sched status:%d", pTask->id.idStr, pTask->status.schedStatus);
}
return 0;
}
int32_t streamTaskSchedTask(SMsgCb* pMsgCb, int32_t vgId, int64_t streamId, int32_t taskId, int32_t execType) {
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));
if (pRunReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
stError("vgId:%d failed to create msg to start stream task:0x%x exec, type:%d, code:%s", vgId, taskId, execType,
terrstr());
return -1;
}
stDebug("vgId:%d create msg to start stream task:0x%x, exec type:%d", vgId, taskId, execType);
pRunReq->head.vgId = vgId;
pRunReq->streamId = streamId;
pRunReq->taskId = taskId;
pRunReq->reqType = execType;
SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
tmsgPutToQueue(pMsgCb, STREAM_QUEUE, &msg);
return TSDB_CODE_SUCCESS;
}
void streamTaskClearSchedIdleInfo(SStreamTask* pTask) { pTask->status.schedIdleTime = 0; }
void streamTaskSetIdleInfo(SStreamTask* pTask, int32_t idleTime) { pTask->status.schedIdleTime = idleTime; }
int32_t streamTaskResumeInFuture(SStreamTask* pTask) {
int32_t ref = atomic_add_fetch_32(&pTask->status.timerActive, 1);
stDebug("s-task:%s task should idle, add into timer to retry in %dms, ref:%d", pTask->id.idStr,
pTask->status.schedIdleTime, ref);
// add one ref count for task
/*SStreamTask* pAddRefTask = */streamMetaAcquireOneTask(pTask);
if (pTask->schedInfo.pIdleTimer == NULL) {
pTask->schedInfo.pIdleTimer = taosTmrStart(streamTaskResumeHelper, pTask->status.schedIdleTime, pTask, streamTimer);
} else {
taosTmrReset(streamTaskResumeHelper, pTask->status.schedIdleTime, pTask, streamTimer, &pTask->schedInfo.pIdleTimer);
}
return TSDB_CODE_SUCCESS;
}
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void streamTaskResumeHelper(void* param, void* tmrId) {
SStreamTask* pTask = (SStreamTask*)param;
SStreamTaskId* pId = &pTask->id;
SStreamTaskState* p = streamTaskGetStatus(pTask);
if (p->state == TASK_STATUS__DROPPING || p->state == TASK_STATUS__STOP) {
streamTaskSetSchedStatusInactive(pTask);
int32_t ref = atomic_sub_fetch_32(&pTask->status.timerActive, 1);
stDebug("s-task:%s status:%s not resume task, ref:%d", pId->idStr, p->name, ref);
streamMetaReleaseTask(pTask->pMeta, pTask);
return;
}
streamTaskSchedTask(pTask->pMsgCb, pTask->info.nodeId, pId->streamId, pId->taskId, STREAM_EXEC_T_RESUME_TASK);
int32_t ref = atomic_sub_fetch_32(&pTask->status.timerActive, 1);
stDebug("trigger to resume s-task:%s after being idled for %dms, ref:%d", pId->idStr, pTask->status.schedIdleTime,
ref);
// release the task ref count
streamTaskClearSchedIdleInfo(pTask);
streamMetaReleaseTask(pTask->pMeta, pTask);
}
void streamSchedByTimer(void* param, void* tmrId) {
SStreamTask* pTask = (void*)param;
const char* id = pTask->id.idStr;
int32_t nextTrigger = (int32_t)pTask->info.triggerParam;
int8_t status = atomic_load_8(&pTask->schedInfo.status);
stTrace("s-task:%s in scheduler, trigger status:%d, next:%dms", id, status, nextTrigger);
if (streamTaskShouldStop(pTask) || streamTaskShouldPause(pTask)) {
stDebug("s-task:%s jump out of schedTimer", id);
return;
}
if (streamTaskGetStatus(pTask)->state == TASK_STATUS__CK) {
stDebug("s-task:%s in checkpoint procedure, not retrieve result, next:%dms", id, nextTrigger);
} else {
if (status == TASK_TRIGGER_STATUS__ACTIVE) {
SStreamTrigger* pTrigger = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0);
if (pTrigger == NULL) {
stError("s-task:%s failed to prepare retrieve data trigger, code:%s, try again in %dms", id, "out of memory",
nextTrigger);
taosTmrReset(streamSchedByTimer, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
return;
}
pTrigger->type = STREAM_INPUT__GET_RES;
pTrigger->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
if (pTrigger->pBlock == NULL) {
taosFreeQitem(pTrigger);
stError("s-task:%s failed to prepare retrieve data trigger, code:%s, try again in %dms", id, "out of memory",
nextTrigger);
taosTmrReset(streamSchedByTimer, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
return;
}
atomic_store_8(&pTask->schedInfo.status, TASK_TRIGGER_STATUS__INACTIVE);
pTrigger->pBlock->info.type = STREAM_GET_ALL;
int32_t code = streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pTrigger);
if (code != TSDB_CODE_SUCCESS) {
taosTmrReset(streamSchedByTimer, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
return;
}
streamTrySchedExec(pTask);
}
}
taosTmrReset(streamSchedByTimer, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
}

52
source/libs/stream/src/streamTask.c
View File

@ -992,3 +992,55 @@ int32_t streamTaskSendCheckpointReq(SStreamTask* pTask) {
tmsgSendReq(&pTask->info.mnodeEpset, &msg); tmsgSendReq(&pTask->info.mnodeEpset, &msg);
return 0; return 0;
} }
SStreamChildEpInfo* streamTaskGetUpstreamTaskEpInfo(SStreamTask* pTask, int32_t taskId) {
int32_t num = taosArrayGetSize(pTask->upstreamInfo.pList);
for (int32_t i = 0; i < num; ++i) {
SStreamChildEpInfo* pInfo = taosArrayGetP(pTask->upstreamInfo.pList, i);
if (pInfo->taskId == taskId) {
return pInfo;
}
}
stError("s-task:%s failed to find upstream task:0x%x", pTask->id.idStr, taskId);
return NULL;
}
char* createStreamTaskIdStr(int64_t streamId, int32_t taskId) {
char buf[128] = {0};
sprintf(buf, "0x%" PRIx64 "-0x%x", streamId, taskId);
return taosStrdup(buf);
}
static int32_t streamTaskEnqueueRetrieve(SStreamTask* pTask, SStreamRetrieveReq* pReq) {
SStreamDataBlock* pData = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SStreamDataBlock));
int8_t status = TASK_INPUT_STATUS__NORMAL;
// enqueue
if (pData != NULL) {
stDebug("s-task:%s (child %d) recv retrieve req from task:0x%x(vgId:%d), reqId:0x%" PRIx64, pTask->id.idStr,
pTask->info.selfChildId, pReq->srcTaskId, pReq->srcNodeId, pReq->reqId);
pData->type = STREAM_INPUT__DATA_RETRIEVE;
pData->srcVgId = 0;
streamRetrieveReqToData(pReq, pData);
if (streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pData) == 0) {
status = TASK_INPUT_STATUS__NORMAL;
} else {
status = TASK_INPUT_STATUS__FAILED;
}
} else { // todo handle oom
/*streamTaskInputFail(pTask);*/
/*status = TASK_INPUT_STATUS__FAILED;*/
}
return status == TASK_INPUT_STATUS__NORMAL ? 0 : -1;
}
int32_t streamProcessRetrieveReq(SStreamTask* pTask, SStreamRetrieveReq* pReq) {
int32_t code = streamTaskEnqueueRetrieve(pTask, pReq);
if(code != 0){
return code;
}
return streamTrySchedExec(pTask);
}

40
source/libs/stream/src/streamTimer.c Normal file
View File

@ -0,0 +1,40 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "streamInt.h"
#include "ttimer.h"
void* streamTimer = NULL;
int32_t streamTimerInit() {
streamTimer = taosTmrInit(1000, 100, 10000, "STREAM");
if (streamTimer == NULL) {
stError("init stream timer failed, code:%s", tstrerror(terrno));
return -1;
}
stInfo("init stream timer, %p", streamTimer);
return 0;
}
void streamTimerCleanUp() {
stInfo("cleanup stream timer, %p", streamTimer);
taosTmrCleanUp(streamTimer);
streamTimer = NULL;
}
tmr_h streamTimerGetInstance() {
return streamTimer;
}