#4984

Committed 13 Mar 2026 03:38AM UTC coverage: 68.643% (-0.01%) from 68.653%

Build # #4984

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat/6641435300-save-audit-in-self (#34738)

Coverage Stats

434 of 584 new or added lines in 10 files covered. (74.32%)

3048 existing lines in 150 files now uncovered.

212713 of 309883 relevant lines covered (68.64%)

135561814.23 hits per line

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

71.01

/source/dnode/vnode/src/tq/tqPush.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 "tq.h"
#include "vnd.h"

int32_t tqProcessSubmitReqForSubscribe(STQ* pTq) {
  if (pTq == NULL) {
    return TSDB_CODE_INVALID_MSG;
  }
  if (taosHashGetSize(pTq->pPushMgr) <= 0) {
    return 0;
  }
  SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME_PUSH};
  msg.pCont = rpcMallocCont(sizeof(SMsgHead));
  if (msg.pCont == NULL) {
    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);
  }
  msg.contLen = sizeof(SMsgHead);
  SMsgHead *pHead = msg.pCont;
  pHead->vgId = TD_VID(pTq->pVnode);
  pHead->contLen = msg.contLen;
  int32_t code = tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg);
  if (code != 0){
    tqError("vgId:%d failed to push msg to queue, code:%d", TD_VID(pTq->pVnode), code);
    rpcFreeCont(msg.pCont);
  }
  return code;
}

int32_t tqPushMsg(STQ* pTq, tmsg_t msgType) {
  int32_t code = 0;
  if (msgType == TDMT_VND_SUBMIT) {
    code = tqProcessSubmitReqForSubscribe(pTq);
    if (code != 0){
      tqError("vgId:%d failed to process submit request for subscribe, code:%d", TD_VID(pTq->pVnode), code);
    }
  }

  return code;
}

int32_t tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg) {
  if (pTq == NULL || handle == NULL || pMsg == NULL) {
    return TSDB_CODE_INVALID_MSG;
  }
  int32_t    vgId = TD_VID(pTq->pVnode);
  STqHandle* pHandle = (STqHandle*)handle;

  if (pHandle->msg == NULL) {
    pHandle->msg = taosMemoryCalloc(1, sizeof(SRpcMsg));
    if (pHandle->msg == NULL) {
      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);
    }
    (void)memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));
    pHandle->msg->pCont = rpcMallocCont(pMsg->contLen);
    if (pHandle->msg->pCont == NULL) {
      taosMemoryFree(pHandle->msg);
      pHandle->msg = NULL;
      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);
    }
  } else {
    tqPushEmptyDataRsp(pHandle, vgId);

    int32_t ret = taosHashRemove(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey));
    tqInfo("vgId:%d register handle, remove last handle from mgr,ret:%s, consumerId:0x%" PRIx64 ", register to pHandle:%p, pCont:%p, len:%d", vgId, tstrerror(ret),
          pHandle->consumerId, pHandle, pHandle->msg->pCont, pHandle->msg->contLen);
    void* tmp = pHandle->msg->pCont;
    (void)memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));
    pHandle->msg->pCont = tmp;
  }

  (void)memcpy(pHandle->msg->pCont, pMsg->pCont, pMsg->contLen);
  pHandle->msg->contLen = pMsg->contLen;
  int32_t ret = taosHashPut(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey), &pHandle, POINTER_BYTES);
  tqInfo("vgId:%d data is over, put handle to mgr,ret:%s, consumerId:0x%" PRIx64 ", register to pHandle:%p, pCont:%p, len:%d", vgId, tstrerror(ret),
          pHandle->consumerId, pHandle, pHandle->msg->pCont, pHandle->msg->contLen);
  if (ret != 0) {
    rpcFreeCont(pHandle->msg->pCont);
    taosMemoryFree(pHandle->msg);
    pHandle->msg = NULL;
  }
  return ret;
}

void tqUnregisterPushHandle(STQ* pTq, void *handle) {
  if (pTq == NULL || handle == NULL) {
    return;
  }
  STqHandle *pHandle = (STqHandle*)handle;
  int32_t    vgId = TD_VID(pTq->pVnode);

  if(taosHashGetSize(pTq->pPushMgr) <= 0) {
    return;
  }
  int32_t ret = taosHashRemove(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey));
  tqInfo("vgId:%d remove pHandle:%p,ret:%d consumer Id:0x%" PRIx64, vgId, pHandle, ret, pHandle->consumerId);

  if(ret == 0 && pHandle->msg != NULL) {
    tqPushEmptyDataRsp(pHandle, vgId);

    rpcFreeCont(pHandle->msg->pCont);
    taosMemoryFree(pHandle->msg);
    pHandle->msg = NULL;
  }
}

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 "tq.h"
17	#include "vnd.h"
18
19	int32_t tqProcessSubmitReqForSubscribe(STQ* pTq) {	705,588,846✔
20	if (pTq == NULL) {	705,588,846✔
21	return TSDB_CODE_INVALID_MSG;	×
22	}
23	if (taosHashGetSize(pTq->pPushMgr) <= 0) {	705,588,846✔
24	return 0;	694,866,539✔
25	}
26	SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME_PUSH};	10,723,056✔
27	msg.pCont = rpcMallocCont(sizeof(SMsgHead));	10,723,056✔
28	if (msg.pCont == NULL) {	10,723,056✔
29	return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);	×
30	}
31	msg.contLen = sizeof(SMsgHead);	10,723,056✔
32	SMsgHead *pHead = msg.pCont;	10,723,056✔
33	pHead->vgId = TD_VID(pTq->pVnode);	10,723,056✔
34	pHead->contLen = msg.contLen;	10,723,056✔
35	int32_t code = tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg);	10,723,056✔
36	if (code != 0){	10,723,056✔
UNCOV 37	tqError("vgId:%d failed to push msg to queue, code:%d", TD_VID(pTq->pVnode), code);	×
UNCOV 38	rpcFreeCont(msg.pCont);	×
39	}
40	return code;	10,723,056✔
41	}
42
43	int32_t tqPushMsg(STQ* pTq, tmsg_t msgType) {	811,368,733✔
44	int32_t code = 0;	811,368,733✔
45	if (msgType == TDMT_VND_SUBMIT) {	811,368,733✔
46	code = tqProcessSubmitReqForSubscribe(pTq);	705,589,962✔
47	if (code != 0){	705,589,372✔
UNCOV 48	tqError("vgId:%d failed to process submit request for subscribe, code:%d", TD_VID(pTq->pVnode), code);	×
49	}
50	}
51
52	return code;	811,370,609✔
53	}
54
55	int32_t tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg) {	11,406,328✔
56	if (pTq == NULL \|\| handle == NULL \|\| pMsg == NULL) {	11,406,328✔
57	return TSDB_CODE_INVALID_MSG;	×
58	}
59	int32_t vgId = TD_VID(pTq->pVnode);	11,406,606✔
60	STqHandle* pHandle = (STqHandle*)handle;	11,406,606✔
61
62	if (pHandle->msg == NULL) {	11,406,606✔
63	pHandle->msg = taosMemoryCalloc(1, sizeof(SRpcMsg));	11,406,606✔
64	if (pHandle->msg == NULL) {	11,406,606✔
65	return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);	×
66	}
67	(void)memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));	11,406,606✔
68	pHandle->msg->pCont = rpcMallocCont(pMsg->contLen);	11,406,606✔
69	if (pHandle->msg->pCont == NULL) {	11,406,606✔
70	taosMemoryFree(pHandle->msg);	×
71	pHandle->msg = NULL;	×
72	return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);	×
73	}
74	} else {
75	tqPushEmptyDataRsp(pHandle, vgId);	×
76
77	int32_t ret = taosHashRemove(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey));	×
78	tqInfo("vgId:%d register handle, remove last handle from mgr,ret:%s, consumerId:0x%" PRIx64 ", register to pHandle:%p, pCont:%p, len:%d", vgId, tstrerror(ret),	×
79	pHandle->consumerId, pHandle, pHandle->msg->pCont, pHandle->msg->contLen);
80	void* tmp = pHandle->msg->pCont;	×
81	(void)memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));	×
82	pHandle->msg->pCont = tmp;	×
83	}
84
85	(void)memcpy(pHandle->msg->pCont, pMsg->pCont, pMsg->contLen);	11,406,606✔
86	pHandle->msg->contLen = pMsg->contLen;	11,406,606✔
87	int32_t ret = taosHashPut(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey), &pHandle, POINTER_BYTES);	11,406,606✔
88	tqInfo("vgId:%d data is over, put handle to mgr,ret:%s, consumerId:0x%" PRIx64 ", register to pHandle:%p, pCont:%p, len:%d", vgId, tstrerror(ret),	11,406,606✔
89	pHandle->consumerId, pHandle, pHandle->msg->pCont, pHandle->msg->contLen);
90	if (ret != 0) {	11,406,606✔
91	rpcFreeCont(pHandle->msg->pCont);	×
92	taosMemoryFree(pHandle->msg);	×
93	pHandle->msg = NULL;	×
94	}
95	return ret;	11,406,606✔
96	}
97
98	void tqUnregisterPushHandle(STQ* pTq, void *handle) {	1,225,211✔
99	if (pTq == NULL \|\| handle == NULL) {	1,225,211✔
100	return;	×
101	}
102	STqHandle pHandle = (STqHandle)handle;	1,225,945✔
103	int32_t vgId = TD_VID(pTq->pVnode);	1,225,945✔
104
105	if(taosHashGetSize(pTq->pPushMgr) <= 0) {	1,225,945✔
106	return;	1,056,662✔
107	}
108	int32_t ret = taosHashRemove(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey));	169,283✔
109	tqInfo("vgId:%d remove pHandle:%p,ret:%d consumer Id:0x%" PRIx64, vgId, pHandle, ret, pHandle->consumerId);	169,283✔
110
111	if(ret == 0 && pHandle->msg != NULL) {	169,283✔
112	tqPushEmptyDataRsp(pHandle, vgId);	166,205✔
113
114	rpcFreeCont(pHandle->msg->pCont);	166,205✔
115	taosMemoryFree(pHandle->msg);	166,205✔
116	pHandle->msg = NULL;	165,883✔
117	}
118	}

taosdata / TDengine / #4984

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