#4513

Committed 17 Jul 2025 02:02AM UTC coverage: 31.359% (-31.1%) from 62.446%

Build # #4513

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #31914 from taosdata/fix/3.0/compare-ans-failed

fix:Convert line endings from LF to CRLF for ans file

Run Details

68541 of 301034 branches covered (22.77%)

Branch coverage included in aggregate %.

117356 of 291771 relevant lines covered (40.22%)

602262.98 hits per line

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

0.0

/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 "tdatablock.h"
#include "tuuid.h"
#include "stream.h"
#include "streamRunner.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->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};
  rpcSendResponse(&rsp);

  streamReleaseTask(taskAddr);

  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));
  int64_t streamId = *(int64_t*)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead) + 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 || (terrno == TAOS_SYSTEM_ERROR(ENOENT) && ver == -1)){
    goto end;
  }
  if (terrno == TAOS_SYSTEM_ERROR(ENOENT) || ver > *(int32_t*)data) {
    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 (terrno == TAOS_SYSTEM_ERROR(ENOENT) || ver >= *(int32_t*)data) {
    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 = INT_BYTES + LONG_BYTES;
    rsp.pCont = rpcMallocCont(rsp.contLen);
    if (rsp.pCont == NULL) {
      rsp.code = TSDB_CODE_OUT_OF_MEMORY;
    } else {
      *(int32_t*)rsp.pCont = -1;  // no checkpoint
      *(int64_t*)(POINTER_SHIFT(rsp.pCont, 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); 
    } 
  }
  
  rpcSendResponse(&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;
  stDebug("[checkpoint] handleSyncWriteCheckPointRsp, dataLen:%d", dataLen);
  
  int32_t ver = *(int32_t*)data;
  int64_t streamId = *(int64_t*)(POINTER_SHIFT(data, INT_BYTES));
  if (ver != -1){
    (void)streamWriteCheckPoint(streamId, data, dataLen);
  }
  return streamCheckpointSetReady(streamId);
}

static int32_t buildFetchRsp(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 : blockGetEncodeSize(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);
    int32_t actualLen = blockEncode(pBlock, pRetrieve->data + INT_BYTES * 2, blockSize, taosArrayGetSize(pBlock->pDataBlock));
    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:%d, contLen:%d", pRpcMsg->msgType, pRpcMsg->contLen);
  
  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;
  calcReq.sessionId = req.pStRtFuncInfo->sessionId;
  calcReq.triggerType = req.pStRtFuncInfo->triggerType;
  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->pMsgCb = &pSnode->msgCb;
  pTask->pWorkerCb = pWorkerCb;
  
  TAOS_CHECK_EXIT(stRunnerTaskExecute(pTask, &calcReq));

  TAOS_CHECK_EXIT(buildFetchRsp(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 = pParam->wstart;
  readInfo.end = pParam->wend;
  bool finished;
  TAOS_CHECK_EXIT(stRunnerFetchDataFromCache(&readInfo,&finished));

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

_exit:

  stsDebug("task %" PRIx64 " TDMT_STREAM_FETCH_FROM_CACHE_RSP with code:%d rows:%" PRId64 ", size:%d", req.taskId, code, readInfo.pBlock ? readInfo.pBlock->info.rows : 0, (int32_t)size);  
  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);
}


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;
    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 "tdatablock.h"
19	#include "tuuid.h"
20	#include "stream.h"
21	#include "streamRunner.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) {	×
29	int32_t code = 0;	×
30	SSnode *pSnode = taosMemoryCalloc(1, sizeof(SSnode));	×
31	if (pSnode == NULL) {	×
32	return NULL;	×
33	}
34
35	pSnode->msgCb = pOption->msgCb;	×
36
37	return pSnode;	×
38	}
39
40	int32_t sndInit(SSnode *pSnode) {	×
41	streamSetSnodeEnabled(&pSnode->msgCb);	×
42	return 0;	×
43	}
44
45	void sndClose(SSnode *pSnode) {	×
46	streamSetSnodeDisabled(false);	×
47	taosMemoryFree(pSnode);	×
48	}	×
49
50	static int32_t handleTriggerCalcReq(SSnode* pSnode, void* pWorkerCb, SRpcMsg* pRpcMsg) {	×
51	SSTriggerCalcRequest req = {0};	×
52	SStreamRunnerTask* pTask = NULL;	×
53	void* taskAddr = NULL;	×
54	int32_t code = 0, lino = 0;	×
55	TAOS_CHECK_EXIT(tDeserializeSTriggerCalcRequest(POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead)), pRpcMsg->contLen - sizeof(SMsgHead), &req));	×
56	TAOS_CHECK_EXIT(streamAcquireTask(req.streamId, req.runnerTaskId, (SStreamTask**)&pTask, &taskAddr));	×
57
58	req.brandNew = true;	×
59	req.execId = -1;	×
60	pTask->pMsgCb = &pSnode->msgCb;	×
61	pTask->pWorkerCb = pWorkerCb;	×
62	req.curWinIdx = 0;	×
63	TAOS_CHECK_EXIT(stRunnerTaskExecute(pTask, &req));	×
64
65	_exit:	×
66
67	tDestroySTriggerCalcRequest(&req);	×
68	SRpcMsg rsp = {.code = code, .msgType = TDMT_STREAM_TRIGGER_CALC_RSP, .contLen = 0, .pCont = NULL, .info = pRpcMsg->info};	×
69	rpcSendResponse(&rsp);	×
70
71	streamReleaseTask(taskAddr);	×
72
73	if (code) {	×
74	sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	×
75	}
76
77	return code;	×
78	}
79
80	static int32_t handleSyncDeleteCheckPointReq(SSnode* pSnode, SRpcMsg* pRpcMsg) {	×
81	int64_t streamId = (int64_t)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead));	×
82	streamDeleteCheckPoint(streamId);	×
83	return 0;	×
84	}
85
86	static int32_t handleSyncWriteCheckPointReq(SSnode* pSnode, SRpcMsg* pRpcMsg) {	×
87	int32_t ver = (int32_t)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead));	×
88	int64_t streamId = (int64_t)POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead) + INT_BYTES);	×
89	SRpcMsg rsp = {.code = 0, .msgType = TDMT_STREAM_SYNC_CHECKPOINT_RSP, .info = pRpcMsg->info};	×
90
91	stDebug("[checkpoint] handleSyncWriteCheckPointReq streamId:%" PRIx64 ",ver:%d", streamId, ver);	×
92	void* data = NULL;	×
93	int64_t dataLen = 0;	×
94	int32_t code = streamReadCheckPoint(streamId, &data, &dataLen);	×
95	if (code != 0 \|\| (terrno == TAOS_SYSTEM_ERROR(ENOENT) && ver == -1)){	×
96	goto end;	×
97	}
98	if (terrno == TAOS_SYSTEM_ERROR(ENOENT) \|\| ver > (int32_t)data) {	×
99	int32_t ret = streamWriteCheckPoint(streamId, POINTER_SHIFT(pRpcMsg->pCont, sizeof(SMsgHead)), pRpcMsg->contLen - sizeof(SMsgHead));	×
100	stDebug("[checkpoint] streamId:%" PRIx64 ", checkpoint local updated, ver:%d, dataLen:%" PRId64 ", ret:%d", streamId, ver, dataLen, ret);	×
101	}
102	if (terrno == TAOS_SYSTEM_ERROR(ENOENT) \|\| ver >= (int32_t)data) {	×
103	stDebug("[checkpoint] streamId:%" PRIx64 ", checkpoint no need send back, ver:%d, dataLen:%" PRId64, streamId, ver, dataLen);	×
104	dataLen = 0;	×
105	taosMemoryFreeClear(data);	×
106	}
107
108	end:	×
109	if (data == NULL) {	×
110	rsp.contLen = INT_BYTES + LONG_BYTES;	×
111	rsp.pCont = rpcMallocCont(rsp.contLen);	×
112	if (rsp.pCont == NULL) {	×
113	rsp.code = TSDB_CODE_OUT_OF_MEMORY;	×
114	} else {
115	(int32_t)rsp.pCont = -1; // no checkpoint	×
116	(int64_t)(POINTER_SHIFT(rsp.pCont, INT_BYTES)) = streamId;	×
117	}
118	} else {
119	rsp.pCont = rpcMallocCont(dataLen);	×
120	if (rsp.pCont == NULL) {	×
121	rsp.code = TSDB_CODE_OUT_OF_MEMORY;	×
122	} else {
123	memcpy(rsp.pCont, data, dataLen);	×
124	rsp.contLen = dataLen;	×
125	taosMemoryFreeClear(data);	×
126	}
127	}
128
129	rpcSendResponse(&rsp);	×
130	return 0;	×
131	}
132
133	static int32_t handleSyncWriteCheckPointRsp(SSnode* pSnode, SRpcMsg* pRpcMsg) {	×
134	if (pRpcMsg->code != 0) {	×
135	stError("[checkpoint] handleSyncWriteCheckPointRsp, code:%d, msgType:%d", pRpcMsg->code, pRpcMsg->msgType);	×
136	return pRpcMsg->code;	×
137	}
138	void* data = pRpcMsg->pCont;	×
139	int32_t dataLen = pRpcMsg->contLen;	×
140	stDebug("[checkpoint] handleSyncWriteCheckPointRsp, dataLen:%d", dataLen);	×
141
142	int32_t ver = (int32_t)data;	×
143	int64_t streamId = (int64_t)(POINTER_SHIFT(data, INT_BYTES));	×
144	if (ver != -1){	×
145	(void)streamWriteCheckPoint(streamId, data, dataLen);	×
146	}
147	return streamCheckpointSetReady(streamId);	×
148	}
149
150	static int32_t buildFetchRsp(SSDataBlock* pBlock, void** data, size_t* size, int8_t precision, bool finished) {	×
151	int32_t code = 0;	×
152	int32_t lino = 0;	×
153	void* buf = NULL;	×
154
155	int32_t blockSize = pBlock == NULL ? 0 : blockGetEncodeSize(pBlock);	×
156	size_t dataEncodeBufSize = sizeof(SRetrieveTableRsp) + INT_BYTES * 2 + blockSize;	×
157	buf = rpcMallocCont(dataEncodeBufSize);	×
158	if (!buf) {	×
159	code = terrno;	×
160	goto end;	×
161	}
162
163	SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;	×
164	pRetrieve->version = 0;	×
165	pRetrieve->precision = precision;	×
166	pRetrieve->compressed = 0;	×
167	((int32_t)(pRetrieve->data)) = blockSize;	×
168	((int32_t)(pRetrieve->data + INT_BYTES)) = blockSize;	×
169	if (pBlock == NULL \|\| pBlock->info.rows == 0) {	×
170	pRetrieve->numOfRows = 0;	×
171	pRetrieve->numOfBlocks = 0;	×
172	pRetrieve->completed = 1;	×
173	} else {
174	pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);	×
175	pRetrieve->numOfBlocks = htonl(1);	×
176	int32_t actualLen = blockEncode(pBlock, pRetrieve->data + INT_BYTES * 2, blockSize, taosArrayGetSize(pBlock->pDataBlock));	×
177	if (actualLen < 0) {	×
178	code = terrno;	×
179	goto end;	×
180	}
181	}
182	if (finished) {	×
183	pRetrieve->completed = 1;	×
184	}
185
186	*data = buf;	×
187	*size = dataEncodeBufSize;	×
188	buf = NULL;	×
189
190	end:	×
191	rpcFreeCont(buf);	×
192	return code;	×
193	}
194
195	static int32_t handleStreamFetchData(SSnode* pSnode, void pWorkerCb, SRpcMsg pRpcMsg) {	×
196	int32_t code = 0, lino = 0;	×
197	void* taskAddr = NULL;	×
198	SResFetchReq req = {0};	×
199	SSTriggerCalcRequest calcReq = {0};	×
200	SStreamRunnerTask* pTask = NULL;	×
201	void* buf = NULL;	×
202	size_t size = 0;	×
203
204	stDebug("handleStreamFetchData, msgType:%d, contLen:%d", pRpcMsg->msgType, pRpcMsg->contLen);	×
205
206	TAOS_CHECK_EXIT(tDeserializeSResFetchReq(pRpcMsg->pCont,pRpcMsg->contLen, &req));	×
207
208	calcReq.streamId = req.queryId;	×
209	calcReq.runnerTaskId = req.taskId;	×
210	calcReq.brandNew = req.reset;	×
211	calcReq.execId = req.execId;	×
212	calcReq.sessionId = req.pStRtFuncInfo->sessionId;	×
213	calcReq.triggerType = req.pStRtFuncInfo->triggerType;	×
214	TSWAP(calcReq.groupColVals, req.pStRtFuncInfo->pStreamPartColVals);	×
215	TSWAP(calcReq.params, req.pStRtFuncInfo->pStreamPesudoFuncVals);	×
216	calcReq.gid = req.pStRtFuncInfo->groupId;	×
217	calcReq.curWinIdx = req.pStRtFuncInfo->curIdx;	×
218	calcReq.pOutBlock = NULL;	×
219
220	TAOS_CHECK_EXIT(streamAcquireTask(calcReq.streamId, calcReq.runnerTaskId, (SStreamTask**)&pTask, &taskAddr));	×
221
222	pTask->pMsgCb = &pSnode->msgCb;	×
223	pTask->pWorkerCb = pWorkerCb;	×
224
225	TAOS_CHECK_EXIT(stRunnerTaskExecute(pTask, &calcReq));	×
226
227	TAOS_CHECK_EXIT(buildFetchRsp(calcReq.pOutBlock, &buf, &size, 0, false));	×
228
229	_exit:	×
230
231	tDestroySTriggerCalcRequest(&calcReq);	×
232	tDestroySResFetchReq(&req);	×
233	SRpcMsg rsp = {.code = code, .msgType = TDMT_STREAM_FETCH_FROM_RUNNER_RSP, .contLen = size, .pCont = buf, .info = pRpcMsg->info};	×
234	tmsgSendRsp(&rsp);	×
235
236	streamReleaseTask(taskAddr);	×
237
238	if (code) {	×
239	sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	×
240	}
241
242	return code;	×
243	}
244
245	static int32_t handleStreamFetchFromCache(SSnode* pSnode, SRpcMsg* pRpcMsg) {	×
246	int32_t code = 0, lino = 0;	×
247	SResFetchReq req = {0};	×
248	SStreamCacheReadInfo readInfo = {0};	×
249	void* buf = NULL;	×
250	int64_t streamId = 0;	×
251	size_t size = 0;	×
252	TAOS_CHECK_EXIT(tDeserializeSResFetchReq(pRpcMsg->pCont, pRpcMsg->contLen, &req));	×
253
254	streamId = req.queryId;	×
255	readInfo.taskInfo.streamId = req.queryId;	×
256	readInfo.taskInfo.taskId = req.taskId;	×
257	readInfo.taskInfo.sessionId = req.pStRtFuncInfo->sessionId;	×
258	readInfo.gid = req.pStRtFuncInfo->groupId;	×
259	SSTriggerCalcParam* pParam = taosArrayGet(req.pStRtFuncInfo->pStreamPesudoFuncVals, req.pStRtFuncInfo->curIdx);	×
260	readInfo.start = pParam->wstart;	×
261	readInfo.end = pParam->wend;	×
262	bool finished;
263	TAOS_CHECK_EXIT(stRunnerFetchDataFromCache(&readInfo,&finished));	×
264
265	TAOS_CHECK_EXIT(buildFetchRsp(readInfo.pBlock, &buf, &size, 0, finished));	×
266
267	_exit:	×
268
269	stsDebug("task %" PRIx64 " TDMT_STREAM_FETCH_FROM_CACHE_RSP with code:%d rows:%" PRId64 ", size:%d", req.taskId, code, readInfo.pBlock ? readInfo.pBlock->info.rows : 0, (int32_t)size);	×
270	SRpcMsg rsp = {.code = code, .msgType = TDMT_STREAM_FETCH_FROM_CACHE_RSP, .contLen = size, .pCont = buf, .info = pRpcMsg->info};	×
271	tmsgSendRsp(&rsp);	×
272
273	if (code) {	×
274	sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	×
275	}
276
277	blockDataDestroy(readInfo.pBlock);	×
278	tDestroySResFetchReq(&req);	×
279
280	return code;	×
281	}
282
283	static void sndSendErrorRrsp(SRpcMsg *pMsg, int32_t errCode) {	×
284	SRpcMsg rspMsg = {0};	×
285
286	rspMsg.info = pMsg->info;	×
287	rspMsg.pCont = NULL;	×
288	rspMsg.contLen = 0;	×
289	rspMsg.code = errCode;	×
290	rspMsg.msgType = pMsg->msgType;	×
291
292	tmsgSendRsp(&rspMsg);	×
293	}	×
294
295
296	int32_t sndProcessStreamMsg(SSnode pSnode, void pWorkerCb, SRpcMsg *pMsg) {	×
297	int32_t code = 0, lino = 0;	×
298	switch (pMsg->msgType) {	×
299	case TDMT_STREAM_TRIGGER_CALC:	×
300	TAOS_CHECK_EXIT(handleTriggerCalcReq(pSnode, pWorkerCb, pMsg));	×
301	break;	×
302	case TDMT_STREAM_DELETE_CHECKPOINT:	×
303	TAOS_CHECK_EXIT(handleSyncDeleteCheckPointReq(pSnode, pMsg));	×
304	break;	×
305	case TDMT_STREAM_SYNC_CHECKPOINT:	×
306	TAOS_CHECK_EXIT(handleSyncWriteCheckPointReq(pSnode, pMsg));	×
307	break;	×
308	case TDMT_STREAM_SYNC_CHECKPOINT_RSP:	×
309	TAOS_CHECK_EXIT(handleSyncWriteCheckPointRsp(pSnode, pMsg));	×
310	break;	×
311	case TDMT_STREAM_FETCH_FROM_RUNNER:	×
312	TAOS_CHECK_EXIT(handleStreamFetchData(pSnode, pWorkerCb, pMsg));	×
313	break;	×
314	case TDMT_STREAM_FETCH_FROM_CACHE:	×
315	TAOS_CHECK_EXIT(handleStreamFetchFromCache(pSnode, pMsg));	×
316	break;	×
317	default:	×
318	sndError("invalid snode msg:%d", pMsg->msgType);	×
319	TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);	×
320	}
321
322	_exit:	×
323
324	if (code) {	×
325	sndError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	×
326	}
327
328	return code;	×
329	}

taosdata / TDengine / #4513

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