#3410

Committed 11 Apr 2024 11:33AM UTC coverage: 67.073% (+0.6%) from 66.463%

Build # #3410

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #25340 from taosdata/fix/TD-29533

fix: arb assigned step down need to reset token

Run Details

93830 of 173326 branches covered (54.13%)

Branch coverage included in aggregate %.

1 of 6 new or added lines in 2 files covered. (16.67%)

1669 existing lines in 114 files now uncovered.

165537 of 213367 relevant lines covered (77.58%)

14248066.43 hits per line

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

87.32

/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 (taosHashGetSize(pTq->pPushMgr) <= 0) {
    return 0;
  }
  SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME_PUSH};
  msg.pCont = rpcMallocCont(sizeof(SMsgHead));
  msg.contLen = sizeof(SMsgHead);
  SMsgHead *pHead = msg.pCont;
  pHead->vgId = TD_VID(pTq->pVnode);
  pHead->contLen = msg.contLen;
  tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg);
  return 0;
}

int32_t tqPushMsg(STQ* pTq, tmsg_t msgType) {
  if (msgType == TDMT_VND_SUBMIT) {
    tqProcessSubmitReqForSubscribe(pTq);
  }

  streamMetaRLock(pTq->pStreamMeta);
  int32_t numOfTasks = streamMetaGetNumOfTasks(pTq->pStreamMeta);
  streamMetaRUnLock(pTq->pStreamMeta);

//  tqTrace("vgId:%d handle submit, restore:%d, numOfTasks:%d", TD_VID(pTq->pVnode), pTq->pVnode->restored, numOfTasks);

  // push data for stream processing:
  // 1. the vnode has already been restored.
  // 2. the vnode should be the leader.
  // 3. the stream is not suspended yet.
  if ((!tsDisableStream) && (numOfTasks > 0) && (msgType == TDMT_VND_SUBMIT || msgType == TDMT_VND_DELETE)) {
    tqScanWalAsync(pTq, true);
  }

  return 0;
}

int32_t tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg) {
  int32_t    vgId = TD_VID(pTq->pVnode);
  STqHandle* pHandle = (STqHandle*)handle;

  if (pHandle->msg == NULL) {
    pHandle->msg = taosMemoryCalloc(1, sizeof(SRpcMsg));
    memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));
    pHandle->msg->pCont = rpcMallocCont(pMsg->contLen);
  } else {
//    tqPushDataRsp(pHandle, vgId);
    tqPushEmptyDataRsp(pHandle, vgId);

    void* tmp = pHandle->msg->pCont;
    memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));
    pHandle->msg->pCont = tmp;
  }

  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);
  tqDebug("vgId:%d data is over, ret:%d, consumerId:0x%" PRIx64 ", register to pHandle:%p, pCont:%p, len:%d", vgId, ret,
          pHandle->consumerId, pHandle, pHandle->msg->pCont, pHandle->msg->contLen);
  return 0;
}

int tqUnregisterPushHandle(STQ* pTq, void *handle) {
  STqHandle *pHandle = (STqHandle*)handle;
  int32_t    vgId = TD_VID(pTq->pVnode);

  if(taosHashGetSize(pTq->pPushMgr) <= 0) {
    return 0;
  }
  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(pHandle->msg != NULL) {
//    tqPushDataRsp(pHandle, vgId);
    tqPushEmptyDataRsp(pHandle, vgId);

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

  return 0;
}

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) {	12,792,586✔
20	if (taosHashGetSize(pTq->pPushMgr) <= 0) {	12,792,586!
21	return 0;	12,792,496✔
22	}
UNCOV 23	SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME_PUSH};	×
UNCOV 24	msg.pCont = rpcMallocCont(sizeof(SMsgHead));	×
25	msg.contLen = sizeof(SMsgHead);	1✔
26	SMsgHead *pHead = msg.pCont;	1✔
27	pHead->vgId = TD_VID(pTq->pVnode);	1✔
28	pHead->contLen = msg.contLen;	1✔
29	tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg);	1✔
30	return 0;	4✔
31	}
32
33	int32_t tqPushMsg(STQ* pTq, tmsg_t msgType) {	12,940,715✔
34	if (msgType == TDMT_VND_SUBMIT) {	12,940,715✔
35	tqProcessSubmitReqForSubscribe(pTq);	12,792,611✔
36	}
37
38	streamMetaRLock(pTq->pStreamMeta);	12,940,602✔
39	int32_t numOfTasks = streamMetaGetNumOfTasks(pTq->pStreamMeta);	12,941,164✔
40	streamMetaRUnLock(pTq->pStreamMeta);	12,940,667✔
41
42	// tqTrace("vgId:%d handle submit, restore:%d, numOfTasks:%d", TD_VID(pTq->pVnode), pTq->pVnode->restored, numOfTasks);
43
44	// push data for stream processing:
45	// 1. the vnode has already been restored.
46	// 2. the vnode should be the leader.
47	// 3. the stream is not suspended yet.
48	if ((!tsDisableStream) && (numOfTasks > 0) && (msgType == TDMT_VND_SUBMIT \|\| msgType == TDMT_VND_DELETE)) {	12,941,249✔
49	tqScanWalAsync(pTq, true);	364,293✔
50	}
51
52	return 0;	12,941,240✔
53	}
54
55	int32_t tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg) {	117✔
56	int32_t vgId = TD_VID(pTq->pVnode);	117✔
57	STqHandle* pHandle = (STqHandle*)handle;	117✔
58
59	if (pHandle->msg == NULL) {	117!
60	pHandle->msg = taosMemoryCalloc(1, sizeof(SRpcMsg));	117✔
61	memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));	117✔
62	pHandle->msg->pCont = rpcMallocCont(pMsg->contLen);	117✔
63	} else {
64	// tqPushDataRsp(pHandle, vgId);
65	tqPushEmptyDataRsp(pHandle, vgId);	×
66
67	void* tmp = pHandle->msg->pCont;	×
68	memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));	×
69	pHandle->msg->pCont = tmp;	×
70	}
71
72	memcpy(pHandle->msg->pCont, pMsg->pCont, pMsg->contLen);	117✔
73	pHandle->msg->contLen = pMsg->contLen;	117✔
74	int32_t ret = taosHashPut(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey), &pHandle, POINTER_BYTES);	117✔
75	tqDebug("vgId:%d data is over, ret:%d, consumerId:0x%" PRIx64 ", register to pHandle:%p, pCont:%p, len:%d", vgId, ret,	117✔
76	pHandle->consumerId, pHandle, pHandle->msg->pCont, pHandle->msg->contLen);
77	return 0;	117✔
78	}
79
80	int tqUnregisterPushHandle(STQ* pTq, void *handle) {	251✔
81	STqHandle pHandle = (STqHandle)handle;	251✔
82	int32_t vgId = TD_VID(pTq->pVnode);	251✔
83
84	if(taosHashGetSize(pTq->pPushMgr) <= 0) {	251✔
85	return 0;	137✔
86	}
87	int32_t ret = taosHashRemove(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey));	114✔
88	tqInfo("vgId:%d remove pHandle:%p,ret:%d consumer Id:0x%" PRIx64, vgId, pHandle, ret, pHandle->consumerId);	114!
89
90	if(pHandle->msg != NULL) {	114✔
91	// tqPushDataRsp(pHandle, vgId);
92	tqPushEmptyDataRsp(pHandle, vgId);	102✔
93
94	rpcFreeCont(pHandle->msg->pCont);	102✔
95	taosMemoryFree(pHandle->msg);	102✔
96	pHandle->msg = NULL;	102✔
97	}
98
99	return 0;	114✔
100	}

taosdata / TDengine / #3410

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