#5059

Committed 17 May 2026 01:15AM UTC coverage: 73.443% (+0.06%) from 73.387%

Build # #5059

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat (TDgpt): Dynamic Model Synchronization Enhancements (#35344)

* refactor: do some internal refactor.

* fix: fix multiprocess sync issue.

* feat: add dynamic anomaly detection and forecasting services

* fix: log error message for undeploying model in exception handling

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix: handle undeploy when model exists only on disk

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/286aafa0-c3ce-4c27-b803-2707571e9dc1

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: guard dynamic registry concurrent access

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: tighten service list locking scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: restore prophet support and update tests per review feedback

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: improve test name and move copy inside lock scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* Potential fix for pull request finding

Co-au... (continued)

Coverage Stats

281870 of 383795 relevant lines covered (73.44%)

135516561.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.87

/source/dnode/snode/src/snode.c

/*
 * 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 "executor.h"
#include "sndInt.h"
#include "stream.h"
#include "streamRunner.h"
#include "tdatablock.h"
#include "tuuid.h"

// clang-format off
#define sndError(...) do {  if (sndDebugFlag & DEBUG_ERROR) { taosPrintLog("SND ERROR ", DEBUG_ERROR, sndDebugFlag, __VA_ARGS__);}} while (0)
#define sndInfo(...)  do {  if (sndDebugFlag & DEBUG_INFO)  { taosPrintLog("SND INFO  ", DEBUG_INFO,  sndDebugFlag, __VA_ARGS__);}} while (0)
#define sndDebug(...) do {  if (sndDebugFlag & DEBUG_DEBUG) { taosPrintLog("SND DEBUG ", DEBUG_DEBUG, sndDebugFlag, __VA_ARGS__);}} while (0)

SSnode *sndOpen(const char *path, const SSnodeOpt *pOption) {
  int32_t code = 0;
  SSnode *pSnode = taosMemoryCalloc(1, sizeof(SSnode));
  if (pSnode == NULL) {
    return NULL;
  }

  pSnode->msgCb = pOption->msgCb;

  return pSnode;
}

int32_t sndInit(SSnode *pSnode) {
  streamSetSnodeEnabled(&pSnode->msgCb);
  return 0;
}

void sndClose(SSnode *pSnode) {
  streamSetSnodeDisabled(false);
  taosMemoryFree(pSnode);
}

static int32_t handleTriggerCalcReq(SSnode* pSnode, void* pWorkerCb, SRpcMsg* pRpcMsg) {
  SSTriggerCalcRequest req = {0};
  SStreamRunnerTask* pTask = NULL;
  void* taskAddr = NULL;
  int32_t code = 0, lino = 0;
  TAOS_CHECK_EXIT(tDeserializeSTriggerCalcRequest(POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead)), pRpcMsg->contLen - sizeof(SMsgHead), &req));
  TAOS_CHECK_EXIT(streamAcquireTask(req.streamId, req.runnerTaskId, (SStreamTask**)&pTask, &taskAddr));

  req.brandNew = true;
  req.execId = -1;
  pTask->msgCb = pSnode->msgCb;
  //pTask->pMsgCb = &pSnode->msgCb;
  pTask->pWorkerCb = pWorkerCb;
  req.curWinIdx = 0;
  TAOS_CHECK_EXIT(stRunnerTaskExecute(pTask, &req));

_exit:

  tDestroySTriggerCalcRequest(&req);
  SRpcMsg rsp = {.code = code, .msgType = TDMT_STREAM_TRIGGER_CALC_RSP, .contLen = 0, .pCont = NULL, .info = pRpcMsg->info};
  if (rpcSendResponse(&rsp) != 0) {
    sndError("failed to send response, msg:%p", &rsp);
  }
  
  streamReleaseTask(taskAddr);

  if(code == TSDB_CODE_MND_STREAM_TABLE_NOT_CREATE) {
    code = 0; // not real error, just notify trigger the table is not created
  }

  if (code) {
    sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));
  }
  
  return code;
}

static int32_t handleSyncDeleteCheckPointReq(SSnode* pSnode, SRpcMsg* pRpcMsg) {
  int64_t streamId = *(int64_t*)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead));
  streamDeleteCheckPoint(streamId);
  return 0;
}

static int32_t handleSyncWriteCheckPointReq(SSnode* pSnode, SRpcMsg* pRpcMsg) {
  int32_t ver = *(int32_t*)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead) + INT_BYTES);
  int64_t streamId = *(int64_t*)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead) + 2 * INT_BYTES);
  SRpcMsg rsp = {.code = 0, .msgType = TDMT_STREAM_SYNC_CHECKPOINT_RSP, .info = pRpcMsg->info};

  stDebug("[checkpoint] handleSyncWriteCheckPointReq streamId:%" PRIx64 ",ver:%d", streamId, ver);
  void*   data = NULL;
  int64_t dataLen = 0;
  int32_t code = streamReadCheckPoint(streamId, &data, &dataLen);
  if (code != 0 || (errorIsFileNotExist(terrno) && ver == -1)) {
    goto end;
  }
  if (errorIsFileNotExist(terrno) || ver > *(int32_t*)POINTER_SHIFT(data, INT_BYTES)) {
    int32_t ret = streamWriteCheckPoint(streamId, POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead)), pRpcMsg->contLen - sizeof(SMsgHead));
    stDebug("[checkpoint] streamId:%" PRIx64 ", checkpoint local updated, ver:%d, dataLen:%" PRId64 ", ret:%d", streamId, ver, dataLen, ret);
  }
  if (errorIsFileNotExist(terrno) || ver >= *(int32_t*)POINTER_SHIFT(data, INT_BYTES)) {
    stDebug("[checkpoint] streamId:%" PRIx64 ", checkpoint no need send back, ver:%d, dataLen:%" PRId64, streamId, ver, dataLen);
    dataLen = 0;
    taosMemoryFreeClear(data);
  }
  
end:
  if (data == NULL) {
    rsp.contLen = 2 * INT_BYTES + LONG_BYTES;
    rsp.pCont = rpcMallocCont(rsp.contLen);
    if (rsp.pCont == NULL) {
      rsp.code = TSDB_CODE_OUT_OF_MEMORY;
    } else {
      *(int32_t*)(POINTER_SHIFT(rsp.pCont, INT_BYTES)) = -1;  // no checkpoint
      *(int64_t*)(POINTER_SHIFT(rsp.pCont, 2 * INT_BYTES)) = streamId;
    }
  } else {
    rsp.pCont = rpcMallocCont(dataLen);
    if (rsp.pCont == NULL) {
      rsp.code = TSDB_CODE_OUT_OF_MEMORY;
    } else {
      memcpy(rsp.pCont, data, dataLen);
      rsp.contLen = dataLen;
      taosMemoryFreeClear(data); 
    } 
  }
  
  if (rpcSendResponse(&rsp) != 0) {
    sndError("failed to send write checkpoint response, msg:%p", &rsp);
  }

  return 0;
}

static int32_t handleSyncWriteCheckPointRsp(SSnode* pSnode, SRpcMsg* pRpcMsg) {
  if (pRpcMsg->code != 0) {
    stError("[checkpoint] handleSyncWriteCheckPointRsp, code:%d, msgType:%d", pRpcMsg->code, pRpcMsg->msgType);
    return pRpcMsg->code;
  } 
  void* data = pRpcMsg->pCont;
  int32_t dataLen = pRpcMsg->contLen;
  
  int32_t ver = *(int32_t*)(POINTER_SHIFT(data, INT_BYTES));
  int64_t streamId = *(int64_t*)(POINTER_SHIFT(data, 2 * INT_BYTES));
  stDebug("[checkpoint] handleSyncWriteCheckPointRsp, ver:%d, streamId:%"PRIx64",dataLen:%d", 
    ver, streamId, dataLen);

  if (ver != -1){
    (void)streamWriteCheckPoint(streamId, data, dataLen);
  }
  return streamCheckpointSetReady(streamId);
}

static int32_t buildStreamFetchRsp(SSDataBlock* pBlock, void** data, size_t* size, int8_t precision, bool finished) {
  int32_t code = 0;
  int32_t lino = 0;
  void*   buf =  NULL;

  int32_t blockSize = pBlock == NULL ? 0 : blockGetInternalEncodeSize(pBlock);
  size_t dataEncodeBufSize = sizeof(SRetrieveTableRsp) + INT_BYTES * 2 + blockSize;
  buf = rpcMallocCont(dataEncodeBufSize);
  if (!buf) {
    code = terrno;
    goto end;
  }

  SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
  pRetrieve->version = 0;
  pRetrieve->precision = precision;
  pRetrieve->compressed = 0;
  *((int32_t*)(pRetrieve->data)) = blockSize;
  *((int32_t*)(pRetrieve->data + INT_BYTES)) = blockSize;
  if (pBlock == NULL || pBlock->info.rows == 0) {
    pRetrieve->numOfRows = 0;
    pRetrieve->numOfBlocks = 0;
    pRetrieve->completed = 1;
  } else {
    pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);
    pRetrieve->numOfBlocks = htonl(1);

    uint32_t numOfCols = (uint32_t)taosArrayGetSize(pBlock->pDataBlock);
    pRetrieve->numOfCols = htonl(numOfCols);

    int32_t actualLen = blockEncodeInternal(pBlock, pRetrieve->data + INT_BYTES * 2, blockSize, numOfCols);
    if (actualLen < 0) {
      code = terrno;
      goto end;
    }
  }
  if (finished) {
    pRetrieve->completed = 1;
  }

  *data = buf;
  *size = dataEncodeBufSize;
  buf = NULL;

end:
  rpcFreeCont(buf);
  return code;
}

static int32_t handleStreamFetchData(SSnode* pSnode, void *pWorkerCb, SRpcMsg* pRpcMsg) {
  int32_t code = 0, lino = 0;
  void* taskAddr = NULL;
  SResFetchReq req = {0};
  SSTriggerCalcRequest calcReq = {0};
  SStreamRunnerTask* pTask = NULL;
  void* buf = NULL;
  size_t size = 0;

  stDebug("handleStreamFetchData, msgType:%s, contLen:%d 0x%" PRIx64 ":0x%" PRIx64, 
      TMSG_INFO(pRpcMsg->msgType), pRpcMsg->contLen, TRACE_GET_ROOTID(&pRpcMsg->info.traceId), TRACE_GET_MSGID(&pRpcMsg->info.traceId));
  
  TAOS_CHECK_EXIT(tDeserializeSResFetchReq(pRpcMsg->pCont,pRpcMsg->contLen, &req));

  calcReq.streamId = req.queryId;
  calcReq.runnerTaskId = req.taskId;
  calcReq.brandNew = req.reset;
  calcReq.execId = req.execId;
  if (req.pStRtFuncInfo) {
    calcReq.sessionId = req.pStRtFuncInfo->sessionId;
    calcReq.triggerType = req.pStRtFuncInfo->triggerType;
    calcReq.isWindowTrigger = req.pStRtFuncInfo->isWindowTrigger;
    calcReq.precision = req.pStRtFuncInfo->precision;
    calcReq.isMultiGroupCalc = req.pStRtFuncInfo->isMultiGroupCalc;
    calcReq.stbPartByTbname = req.pStRtFuncInfo->stbPartByTbname;
    if (calcReq.isMultiGroupCalc) {
      TSWAP(calcReq.pGroupCalcInfos, req.pStRtFuncInfo->pGroupCalcInfos);
      TSWAP(calcReq.pGroupReadInfos, req.pStRtFuncInfo->pGroupReadInfos);
    } else {
      TSWAP(calcReq.groupColVals, req.pStRtFuncInfo->pStreamPartColVals);
      TSWAP(calcReq.params, req.pStRtFuncInfo->pStreamPesudoFuncVals);
      calcReq.gid = req.pStRtFuncInfo->groupId;
    }
    calcReq.curWinIdx = req.pStRtFuncInfo->curIdx;
  }
  calcReq.pOutBlock = NULL;

  TAOS_CHECK_EXIT(streamAcquireTask(calcReq.streamId, calcReq.runnerTaskId, (SStreamTask**)&pTask, &taskAddr));

  pTask->msgCb = pSnode->msgCb;
  //pTask->pMsgCb = &pSnode->msgCb;
  pTask->pWorkerCb = pWorkerCb;
  
  TAOS_CHECK_EXIT(stRunnerTaskExecute(pTask, &calcReq));

  TAOS_CHECK_EXIT(buildStreamFetchRsp(calcReq.pOutBlock, &buf, &size, 0, false));

_exit:

  tDestroySTriggerCalcRequest(&calcReq);
  tDestroySResFetchReq(&req);
  SRpcMsg rsp = {.code = code, .msgType = TDMT_STREAM_FETCH_FROM_RUNNER_RSP, .contLen = size, .pCont = buf, .info = pRpcMsg->info};
  tmsgSendRsp(&rsp);
  
  streamReleaseTask(taskAddr);

  if (code) {
    sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));
  }
  
  return code;
}

static int32_t handleStreamFetchFromCache(SSnode* pSnode, SRpcMsg* pRpcMsg) {
  int32_t code = 0, lino = 0;
  SResFetchReq req = {0};
  SStreamCacheReadInfo readInfo = {0};
  void* buf = NULL;
  int64_t streamId = 0;
  size_t size = 0;
  TAOS_CHECK_EXIT(tDeserializeSResFetchReq(pRpcMsg->pCont, pRpcMsg->contLen, &req));

  streamId = req.queryId;
  readInfo.taskInfo.streamId = req.queryId;
  readInfo.taskInfo.taskId = req.taskId;
  readInfo.taskInfo.sessionId = req.pStRtFuncInfo->sessionId;
  readInfo.gid = req.pStRtFuncInfo->groupId;
  //SSTriggerCalcParam* pParam = taosArrayGet(req.pStRtFuncInfo->pStreamPesudoFuncVals, req.pStRtFuncInfo->curIdx);
  readInfo.start = req.pStRtFuncInfo->curWindow.skey;
  readInfo.end = req.pStRtFuncInfo->curWindow.ekey;
  bool finished;
  TAOS_CHECK_EXIT(stRunnerFetchDataFromCache(&readInfo,&finished));

  TAOS_CHECK_EXIT(buildStreamFetchRsp(readInfo.pBlock, &buf, &size, 0, finished));

_exit:

  printDataBlock(readInfo.pBlock, __func__, "fetchFromCache", streamId);

  stsDebug("task %" PRIx64 " TDMT_STREAM_FETCH_FROM_CACHE_RSP with code:%d rows:%" PRId64 ", size:%d, time range:[%" PRId64 ", %" PRId64 "]", 
      req.taskId, code, readInfo.pBlock ? readInfo.pBlock->info.rows : 0, (int32_t)size, readInfo.start, readInfo.end);  
      
  SRpcMsg rsp = {.code = code, .msgType = TDMT_STREAM_FETCH_FROM_CACHE_RSP, .contLen = size, .pCont = buf, .info = pRpcMsg->info};
  tmsgSendRsp(&rsp);

  if (code) {
    sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));
  }

  blockDataDestroy(readInfo.pBlock);
  tDestroySResFetchReq(&req);
  
  return code;
}

static void sndSendErrorRrsp(SRpcMsg *pMsg, int32_t errCode) {
  SRpcMsg             rspMsg = {0};

  rspMsg.info = pMsg->info;
  rspMsg.pCont = NULL;
  rspMsg.contLen = 0;
  rspMsg.code = errCode;
  rspMsg.msgType = pMsg->msgType;

  tmsgSendRsp(&rspMsg);
}

static int32_t handleStreamDropTableReq(SSnode* pSnode, SRpcMsg* pRpcMsg) {
  SSTriggerDropRequest req = {0};
  SStreamRunnerTask* pTask = NULL;
  void* taskAddr = NULL;
  int32_t code = 0, lino = 0;
  TAOS_CHECK_EXIT(tDeserializeSTriggerDropTableRequest(POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead)), pRpcMsg->contLen - sizeof(SMsgHead), &req));
  TAOS_CHECK_EXIT(streamAcquireTask(req.streamId, req.runnerTaskId, (SStreamTask**)&pTask, &taskAddr));
  
  pTask->msgCb = pSnode->msgCb;
  TAOS_CHECK_EXIT(stRunnerTaskDropTable(pTask, &req));

_exit:
  tDestroySSTriggerDropRequest(&req);
  if (code) {
    sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));
    sndSendErrorRrsp(pRpcMsg, code);
  } else {
    SRpcMsg rsp = {.code = 0, .msgType = TDMT_STREAM_TRIGGER_DROP_RSP, .contLen = 0, .pCont = NULL, .info = pRpcMsg->info};
    tmsgSendRsp(&rsp);
  }
  streamReleaseTask(taskAddr);

  return code;
}


int32_t sndProcessStreamMsg(SSnode *pSnode, void *pWorkerCb, SRpcMsg *pMsg) {
  int32_t code = 0, lino = 0;
  switch (pMsg->msgType) {
    case TDMT_STREAM_TRIGGER_CALC:
      TAOS_CHECK_EXIT(handleTriggerCalcReq(pSnode, pWorkerCb, pMsg));
      break;
    case TDMT_STREAM_DELETE_CHECKPOINT:
      TAOS_CHECK_EXIT(handleSyncDeleteCheckPointReq(pSnode, pMsg));
      break;
    case TDMT_STREAM_SYNC_CHECKPOINT:
      TAOS_CHECK_EXIT(handleSyncWriteCheckPointReq(pSnode, pMsg));
      break;
    case TDMT_STREAM_SYNC_CHECKPOINT_RSP:
      TAOS_CHECK_EXIT(handleSyncWriteCheckPointRsp(pSnode, pMsg));
      break;
    case TDMT_STREAM_FETCH_FROM_RUNNER:
      TAOS_CHECK_EXIT(handleStreamFetchData(pSnode, pWorkerCb, pMsg));
      break;
    case TDMT_STREAM_FETCH_FROM_CACHE:
      TAOS_CHECK_EXIT(handleStreamFetchFromCache(pSnode, pMsg));
      break;
      case TDMT_STREAM_TRIGGER_DROP:
     TAOS_CHECK_EXIT(handleStreamDropTableReq(pSnode, pMsg));
      break;
    default:
      sndError("invalid snode msg:%d", pMsg->msgType);
      TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);
  }

_exit:

  if (code) {
    sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));
  }
  
  return code;
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "executor.h"
17	#include "sndInt.h"
18	#include "stream.h"
19	#include "streamRunner.h"
20	#include "tdatablock.h"
21	#include "tuuid.h"
22
23	// clang-format off
24	#define sndError(...) do { if (sndDebugFlag & DEBUG_ERROR) { taosPrintLog("SND ERROR ", DEBUG_ERROR, sndDebugFlag, __VA_ARGS__);}} while (0)
25	#define sndInfo(...) do { if (sndDebugFlag & DEBUG_INFO) { taosPrintLog("SND INFO ", DEBUG_INFO, sndDebugFlag, __VA_ARGS__);}} while (0)
26	#define sndDebug(...) do { if (sndDebugFlag & DEBUG_DEBUG) { taosPrintLog("SND DEBUG ", DEBUG_DEBUG, sndDebugFlag, __VA_ARGS__);}} while (0)
27
28	SSnode sndOpen(const char path, const SSnodeOpt *pOption) {	91,942✔
29	int32_t code = 0;	91,942✔
30	SSnode *pSnode = taosMemoryCalloc(1, sizeof(SSnode));	91,942✔
31	if (pSnode == NULL) {	91,942✔
32	return NULL;	×
33	}
34
35	pSnode->msgCb = pOption->msgCb;	91,942✔
36
37	return pSnode;	91,942✔
38	}
39
40	int32_t sndInit(SSnode *pSnode) {	91,942✔
41	streamSetSnodeEnabled(&pSnode->msgCb);	91,942✔
42	return 0;	91,942✔
43	}
44
45	void sndClose(SSnode *pSnode) {	91,942✔
46	streamSetSnodeDisabled(false);	91,942✔
47	taosMemoryFree(pSnode);	91,942✔
48	}	91,942✔
49
50	static int32_t handleTriggerCalcReq(SSnode* pSnode, void* pWorkerCb, SRpcMsg* pRpcMsg) {	3,194,180✔
51	SSTriggerCalcRequest req = {0};	3,194,180✔
52	SStreamRunnerTask* pTask = NULL;	3,194,180✔
53	void* taskAddr = NULL;	3,194,180✔
54	int32_t code = 0, lino = 0;	3,194,180✔
55	TAOS_CHECK_EXIT(tDeserializeSTriggerCalcRequest(POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead)), pRpcMsg->contLen - sizeof(SMsgHead), &req));	3,194,180✔
56	TAOS_CHECK_EXIT(streamAcquireTask(req.streamId, req.runnerTaskId, (SStreamTask**)&pTask, &taskAddr));	3,193,924✔
57
58	req.brandNew = true;	3,194,180✔
59	req.execId = -1;	3,194,180✔
60	pTask->msgCb = pSnode->msgCb;	3,194,180✔
61	//pTask->pMsgCb = &pSnode->msgCb;
62	pTask->pWorkerCb = pWorkerCb;	3,194,180✔
63	req.curWinIdx = 0;	3,194,180✔
64	TAOS_CHECK_EXIT(stRunnerTaskExecute(pTask, &req));	3,194,180✔
65
66	_exit:	3,194,180✔
67
68	tDestroySTriggerCalcRequest(&req);	3,194,180✔
69	SRpcMsg rsp = {.code = code, .msgType = TDMT_STREAM_TRIGGER_CALC_RSP, .contLen = 0, .pCont = NULL, .info = pRpcMsg->info};	3,194,180✔
70	if (rpcSendResponse(&rsp) != 0) {	3,194,180✔
71	sndError("failed to send response, msg:%p", &rsp);	×
72	}
73
74	streamReleaseTask(taskAddr);	3,193,412✔
75
76	if(code == TSDB_CODE_MND_STREAM_TABLE_NOT_CREATE) {	3,193,924✔
77	code = 0; // not real error, just notify trigger the table is not created	×
78	}
79
80	if (code) {	3,193,924✔
81	sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	1,798,578✔
82	}
83
84	return code;	3,194,180✔
85	}
86
87	static int32_t handleSyncDeleteCheckPointReq(SSnode* pSnode, SRpcMsg* pRpcMsg) {	1,733✔
88	int64_t streamId = (int64_t)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead));	1,733✔
89	streamDeleteCheckPoint(streamId);	1,733✔
90	return 0;	1,733✔
91	}
92
93	static int32_t handleSyncWriteCheckPointReq(SSnode* pSnode, SRpcMsg* pRpcMsg) {	8,253✔
94	int32_t ver = (int32_t)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead) + INT_BYTES);	8,253✔
95	int64_t streamId = (int64_t)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead) + 2 * INT_BYTES);	8,253✔
96	SRpcMsg rsp = {.code = 0, .msgType = TDMT_STREAM_SYNC_CHECKPOINT_RSP, .info = pRpcMsg->info};	8,253✔
97
98	stDebug("[checkpoint] handleSyncWriteCheckPointReq streamId:%" PRIx64 ",ver:%d", streamId, ver);	8,253✔
99	void* data = NULL;	8,253✔
100	int64_t dataLen = 0;	8,253✔
101	int32_t code = streamReadCheckPoint(streamId, &data, &dataLen);	8,253✔
102	if (code != 0 \|\| (errorIsFileNotExist(terrno) && ver == -1)) {	8,253✔
103	goto end;	4,932✔
104	}
105	if (errorIsFileNotExist(terrno) \|\| ver > (int32_t)POINTER_SHIFT(data, INT_BYTES)) {	3,321✔
106	int32_t ret = streamWriteCheckPoint(streamId, POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead)), pRpcMsg->contLen - sizeof(SMsgHead));	2,214✔
107	stDebug("[checkpoint] streamId:%" PRIx64 ", checkpoint local updated, ver:%d, dataLen:%" PRId64 ", ret:%d", streamId, ver, dataLen, ret);	2,214✔
108	}
109	if (errorIsFileNotExist(terrno) \|\| ver >= (int32_t)POINTER_SHIFT(data, INT_BYTES)) {	3,321✔
110	stDebug("[checkpoint] streamId:%" PRIx64 ", checkpoint no need send back, ver:%d, dataLen:%" PRId64, streamId, ver, dataLen);	3,321✔
111	dataLen = 0;	3,321✔
112	taosMemoryFreeClear(data);	3,321✔
113	}
114
115	end:	8,253✔
116	if (data == NULL) {	8,253✔
117	rsp.contLen = 2 * INT_BYTES + LONG_BYTES;	8,253✔
118	rsp.pCont = rpcMallocCont(rsp.contLen);	8,253✔
119	if (rsp.pCont == NULL) {	8,253✔
120	rsp.code = TSDB_CODE_OUT_OF_MEMORY;	×
121	} else {
122	(int32_t)(POINTER_SHIFT(rsp.pCont, INT_BYTES)) = -1; // no checkpoint	8,253✔
123	(int64_t)(POINTER_SHIFT(rsp.pCont, 2 * INT_BYTES)) = streamId;	8,253✔
124	}
125	} else {
126	rsp.pCont = rpcMallocCont(dataLen);	×
127	if (rsp.pCont == NULL) {	×
128	rsp.code = TSDB_CODE_OUT_OF_MEMORY;	×
129	} else {
130	memcpy(rsp.pCont, data, dataLen);	×
131	rsp.contLen = dataLen;	×
132	taosMemoryFreeClear(data);	×
133	}
134	}
135
136	if (rpcSendResponse(&rsp) != 0) {	8,253✔
137	sndError("failed to send write checkpoint response, msg:%p", &rsp);	×
138	}
139
140	return 0;	8,253✔
141	}
142
143	static int32_t handleSyncWriteCheckPointRsp(SSnode* pSnode, SRpcMsg* pRpcMsg) {	8,255✔
144	if (pRpcMsg->code != 0) {	8,255✔
145	stError("[checkpoint] handleSyncWriteCheckPointRsp, code:%d, msgType:%d", pRpcMsg->code, pRpcMsg->msgType);	×
146	return pRpcMsg->code;	×
147	}
148	void* data = pRpcMsg->pCont;	8,255✔
149	int32_t dataLen = pRpcMsg->contLen;	8,255✔
150
151	int32_t ver = (int32_t)(POINTER_SHIFT(data, INT_BYTES));	8,255✔
152	int64_t streamId = (int64_t)(POINTER_SHIFT(data, 2 * INT_BYTES));	8,255✔
153	stDebug("[checkpoint] handleSyncWriteCheckPointRsp, ver:%d, streamId:%"PRIx64",dataLen:%d",	8,255✔
154	ver, streamId, dataLen);
155
156	if (ver != -1){	8,255✔
157	(void)streamWriteCheckPoint(streamId, data, dataLen);	×
158	}
159	return streamCheckpointSetReady(streamId);	8,255✔
160	}
161
162	static int32_t buildStreamFetchRsp(SSDataBlock* pBlock, void** data, size_t* size, int8_t precision, bool finished) {	8,588,432✔
163	int32_t code = 0;	8,588,432✔
164	int32_t lino = 0;	8,588,432✔
165	void* buf = NULL;	8,588,432✔
166
167	int32_t blockSize = pBlock == NULL ? 0 : blockGetInternalEncodeSize(pBlock);	8,588,432✔
168	size_t dataEncodeBufSize = sizeof(SRetrieveTableRsp) + INT_BYTES * 2 + blockSize;	8,588,432✔
169	buf = rpcMallocCont(dataEncodeBufSize);	8,588,432✔
170	if (!buf) {	8,588,432✔
171	code = terrno;	×
172	goto end;	×
173	}
174
175	SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;	8,588,432✔
176	pRetrieve->version = 0;	8,588,432✔
177	pRetrieve->precision = precision;	8,588,432✔
178	pRetrieve->compressed = 0;	8,588,163✔
179	((int32_t)(pRetrieve->data)) = blockSize;	8,588,432✔
180	((int32_t)(pRetrieve->data + INT_BYTES)) = blockSize;	8,587,340✔
181	if (pBlock == NULL \|\| pBlock->info.rows == 0) {	8,587,340✔
182	pRetrieve->numOfRows = 0;	3,726,406✔
183	pRetrieve->numOfBlocks = 0;	3,727,229✔
184	pRetrieve->completed = 1;	3,727,229✔
185	} else {
186	pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);	4,860,934✔
187	pRetrieve->numOfBlocks = htonl(1);	4,861,203✔
188
189	uint32_t numOfCols = (uint32_t)taosArrayGetSize(pBlock->pDataBlock);	4,861,203✔
190	pRetrieve->numOfCols = htonl(numOfCols);	4,860,934✔
191
192	int32_t actualLen = blockEncodeInternal(pBlock, pRetrieve->data + INT_BYTES * 2, blockSize, numOfCols);	4,860,934✔
193	if (actualLen < 0) {	4,861,203✔
194	code = terrno;	×
195	goto end;	×
196	}
197	}
198	if (finished) {	8,588,432✔
199	pRetrieve->completed = 1;	3,718,348✔
200	}
201
202	*data = buf;	8,588,432✔
203	*size = dataEncodeBufSize;	8,588,153✔
204	buf = NULL;	8,587,884✔
205
206	end:	8,587,884✔
207	rpcFreeCont(buf);	8,587,884✔
208	return code;	8,588,432✔
209	}
210
211	static int32_t handleStreamFetchData(SSnode* pSnode, void pWorkerCb, SRpcMsg pRpcMsg) {	3,014,122✔
212	int32_t code = 0, lino = 0;	3,014,122✔
213	void* taskAddr = NULL;	3,014,122✔
214	SResFetchReq req = {0};	3,014,122✔
215	SSTriggerCalcRequest calcReq = {0};	3,014,122✔
216	SStreamRunnerTask* pTask = NULL;	3,014,122✔
217	void* buf = NULL;	3,014,122✔
218	size_t size = 0;	3,014,122✔
219
220	stDebug("handleStreamFetchData, msgType:%s, contLen:%d 0x%" PRIx64 ":0x%" PRIx64,	3,014,122✔
221	TMSG_INFO(pRpcMsg->msgType), pRpcMsg->contLen, TRACE_GET_ROOTID(&pRpcMsg->info.traceId), TRACE_GET_MSGID(&pRpcMsg->info.traceId));
222
223	TAOS_CHECK_EXIT(tDeserializeSResFetchReq(pRpcMsg->pCont,pRpcMsg->contLen, &req));	3,014,122✔
224
225	calcReq.streamId = req.queryId;	3,013,843✔
226	calcReq.runnerTaskId = req.taskId;	3,013,843✔
227	calcReq.brandNew = req.reset;	3,013,843✔
228	calcReq.execId = req.execId;	3,013,843✔
229	if (req.pStRtFuncInfo) {	3,013,843✔
230	calcReq.sessionId = req.pStRtFuncInfo->sessionId;	3,013,843✔
231	calcReq.triggerType = req.pStRtFuncInfo->triggerType;	3,013,843✔
232	calcReq.isWindowTrigger = req.pStRtFuncInfo->isWindowTrigger;	3,013,843✔
233	calcReq.precision = req.pStRtFuncInfo->precision;	3,013,843✔
234	calcReq.isMultiGroupCalc = req.pStRtFuncInfo->isMultiGroupCalc;	3,013,843✔
235	calcReq.stbPartByTbname = req.pStRtFuncInfo->stbPartByTbname;	3,013,843✔
236	if (calcReq.isMultiGroupCalc) {	3,013,843✔
237	TSWAP(calcReq.pGroupCalcInfos, req.pStRtFuncInfo->pGroupCalcInfos);	×
238	TSWAP(calcReq.pGroupReadInfos, req.pStRtFuncInfo->pGroupReadInfos);	×
239	} else {
240	TSWAP(calcReq.groupColVals, req.pStRtFuncInfo->pStreamPartColVals);	3,013,843✔
241	TSWAP(calcReq.params, req.pStRtFuncInfo->pStreamPesudoFuncVals);	3,014,122✔
242	calcReq.gid = req.pStRtFuncInfo->groupId;	3,013,843✔
243	}
244	calcReq.curWinIdx = req.pStRtFuncInfo->curIdx;	3,014,122✔
245	}
246	calcReq.pOutBlock = NULL;	3,013,843✔
247
248	TAOS_CHECK_EXIT(streamAcquireTask(calcReq.streamId, calcReq.runnerTaskId, (SStreamTask**)&pTask, &taskAddr));	3,013,843✔
249
250	pTask->msgCb = pSnode->msgCb;	3,013,843✔
251	//pTask->pMsgCb = &pSnode->msgCb;
252	pTask->pWorkerCb = pWorkerCb;	3,014,122✔
253
254	TAOS_CHECK_EXIT(stRunnerTaskExecute(pTask, &calcReq));	3,013,560✔
255
256	TAOS_CHECK_EXIT(buildStreamFetchRsp(calcReq.pOutBlock, &buf, &size, 0, false));	3,014,122✔
257
258	_exit:	3,013,843✔
259
260	tDestroySTriggerCalcRequest(&calcReq);	3,013,843✔
261	tDestroySResFetchReq(&req);	3,013,843✔
262	SRpcMsg rsp = {.code = code, .msgType = TDMT_STREAM_FETCH_FROM_RUNNER_RSP, .contLen = size, .pCont = buf, .info = pRpcMsg->info};	3,014,122✔
263	tmsgSendRsp(&rsp);	3,014,122✔
264
265	streamReleaseTask(taskAddr);	3,014,122✔
266
267	if (code) {	3,014,122✔
268	sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	×
269	}
270
271	return code;	3,014,122✔
272	}
273
274	static int32_t handleStreamFetchFromCache(SSnode* pSnode, SRpcMsg* pRpcMsg) {	5,574,310✔
275	int32_t code = 0, lino = 0;	5,574,310✔
276	SResFetchReq req = {0};	5,574,310✔
277	SStreamCacheReadInfo readInfo = {0};	5,574,310✔
278	void* buf = NULL;	5,574,310✔
279	int64_t streamId = 0;	5,574,310✔
280	size_t size = 0;	5,574,310✔
281	TAOS_CHECK_EXIT(tDeserializeSResFetchReq(pRpcMsg->pCont, pRpcMsg->contLen, &req));	5,574,310✔
282
283	streamId = req.queryId;	5,573,487✔
284	readInfo.taskInfo.streamId = req.queryId;	5,573,487✔
285	readInfo.taskInfo.taskId = req.taskId;	5,573,487✔
286	readInfo.taskInfo.sessionId = req.pStRtFuncInfo->sessionId;	5,573,487✔
287	readInfo.gid = req.pStRtFuncInfo->groupId;	5,573,219✔
288	//SSTriggerCalcParam* pParam = taosArrayGet(req.pStRtFuncInfo->pStreamPesudoFuncVals, req.pStRtFuncInfo->curIdx);
289	readInfo.start = req.pStRtFuncInfo->curWindow.skey;	5,574,310✔
290	readInfo.end = req.pStRtFuncInfo->curWindow.ekey;	5,572,934✔
291	bool finished;	5,573,126✔
292	TAOS_CHECK_EXIT(stRunnerFetchDataFromCache(&readInfo,&finished));	5,573,487✔
293
294	TAOS_CHECK_EXIT(buildStreamFetchRsp(readInfo.pBlock, &buf, &size, 0, finished));	5,574,310✔
295
296	_exit:	5,574,310✔
297
298	printDataBlock(readInfo.pBlock, __func__, "fetchFromCache", streamId);	5,574,310✔
299
300	stsDebug("task %" PRIx64 " TDMT_STREAM_FETCH_FROM_CACHE_RSP with code:%d rows:%" PRId64 ", size:%d, time range:[%" PRId64 ", %" PRId64 "]",	5,574,041✔
301	req.taskId, code, readInfo.pBlock ? readInfo.pBlock->info.rows : 0, (int32_t)size, readInfo.start, readInfo.end);
302
303	SRpcMsg rsp = {.code = code, .msgType = TDMT_STREAM_FETCH_FROM_CACHE_RSP, .contLen = size, .pCont = buf, .info = pRpcMsg->info};	5,574,864✔
304	tmsgSendRsp(&rsp);	5,574,310✔
305
306	if (code) {	5,574,310✔
307	sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	×
308	}
309
310	blockDataDestroy(readInfo.pBlock);	5,574,310✔
311	tDestroySResFetchReq(&req);	5,574,310✔
312
313	return code;	5,574,310✔
314	}
315
316	static void sndSendErrorRrsp(SRpcMsg *pMsg, int32_t errCode) {	×
317	SRpcMsg rspMsg = {0};	×
318
319	rspMsg.info = pMsg->info;	×
320	rspMsg.pCont = NULL;	×
321	rspMsg.contLen = 0;	×
322	rspMsg.code = errCode;	×
323	rspMsg.msgType = pMsg->msgType;	×
324
325	tmsgSendRsp(&rspMsg);	×
326	}	×
327
328	static int32_t handleStreamDropTableReq(SSnode* pSnode, SRpcMsg* pRpcMsg) {	×
329	SSTriggerDropRequest req = {0};	×
330	SStreamRunnerTask* pTask = NULL;	×
331	void* taskAddr = NULL;	×
332	int32_t code = 0, lino = 0;	×
333	TAOS_CHECK_EXIT(tDeserializeSTriggerDropTableRequest(POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead)), pRpcMsg->contLen - sizeof(SMsgHead), &req));	×
334	TAOS_CHECK_EXIT(streamAcquireTask(req.streamId, req.runnerTaskId, (SStreamTask**)&pTask, &taskAddr));	×
335
336	pTask->msgCb = pSnode->msgCb;	×
337	TAOS_CHECK_EXIT(stRunnerTaskDropTable(pTask, &req));	×
338
339	_exit:	×
340	tDestroySSTriggerDropRequest(&req);	×
341	if (code) {	×
342	sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	×
343	sndSendErrorRrsp(pRpcMsg, code);	×
344	} else {
345	SRpcMsg rsp = {.code = 0, .msgType = TDMT_STREAM_TRIGGER_DROP_RSP, .contLen = 0, .pCont = NULL, .info = pRpcMsg->info};	×
346	tmsgSendRsp(&rsp);	×
347	}
348	streamReleaseTask(taskAddr);	×
349
350	return code;	×
351	}
352
353
354	int32_t sndProcessStreamMsg(SSnode pSnode, void pWorkerCb, SRpcMsg *pMsg) {	11,800,332✔
355	int32_t code = 0, lino = 0;	11,800,332✔
356	switch (pMsg->msgType) {	11,800,332✔
357	case TDMT_STREAM_TRIGGER_CALC:	3,194,180✔
358	TAOS_CHECK_EXIT(handleTriggerCalcReq(pSnode, pWorkerCb, pMsg));	3,194,180✔
359	break;	1,396,425✔
360	case TDMT_STREAM_DELETE_CHECKPOINT:	1,733✔
361	TAOS_CHECK_EXIT(handleSyncDeleteCheckPointReq(pSnode, pMsg));	1,733✔
362	break;	1,733✔
363	case TDMT_STREAM_SYNC_CHECKPOINT:	8,253✔
364	TAOS_CHECK_EXIT(handleSyncWriteCheckPointReq(pSnode, pMsg));	8,253✔
365	break;	8,253✔
366	case TDMT_STREAM_SYNC_CHECKPOINT_RSP:	8,255✔
367	TAOS_CHECK_EXIT(handleSyncWriteCheckPointRsp(pSnode, pMsg));	8,255✔
368	break;	5,156✔
369	case TDMT_STREAM_FETCH_FROM_RUNNER:	3,014,122✔
370	TAOS_CHECK_EXIT(handleStreamFetchData(pSnode, pWorkerCb, pMsg));	3,014,122✔
371	break;	3,014,122✔
372	case TDMT_STREAM_FETCH_FROM_CACHE:	5,574,310✔
373	TAOS_CHECK_EXIT(handleStreamFetchFromCache(pSnode, pMsg));	5,574,310✔
374	break;	5,574,310✔
375	case TDMT_STREAM_TRIGGER_DROP:	×
376	TAOS_CHECK_EXIT(handleStreamDropTableReq(pSnode, pMsg));	×
377	break;	×
378	default:	×
379	sndError("invalid snode msg:%d", pMsg->msgType);	×
380	TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);	×
381	}
382
383	_exit:	×
384
385	if (code) {	11,800,853✔
386	sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	1,801,677✔
387	}
388
389	return code;	11,800,853✔
390	}

taosdata / TDengine / #5059

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous