#3819

Committed 01 Apr 2025 09:27AM UTC coverage: 34.076% (+0.01%) from 34.065%

Build # #3819

Build Type

push

travis-ci

Committed by happyguoxy

Commit Message

test:alter gcda dir

Run Details

148544 of 599532 branches covered (24.78%)

Branch coverage included in aggregate %.

222541 of 489451 relevant lines covered (45.47%)

763329.1 hits per line

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

6.76

/source/dnode/vnode/src/sma/smaCommit.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 "sma.h"

extern SSmaMgmt smaMgmt;

static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit);
static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo);
static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma);
static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pRSmaStat);

/**
 * @brief only applicable to Rollup SMA
 *
 * @param pSma
 * @return int32_t
 */
int32_t smaPreClose(SSma *pSma) { return tdProcessRSmaAsyncPreCommitImpl(pSma, false); }

/**
 * @brief async commit, only applicable to Rollup SMA
 *
 * @param pSma
 * @return int32_t
 */
int32_t smaPrepareAsyncCommit(SSma *pSma) { return tdProcessRSmaAsyncPreCommitImpl(pSma, true); }

/**
 * @brief async commit, only applicable to Rollup SMA
 *
 * @param pSma
 * @return int32_t
 */
int32_t smaCommit(SSma *pSma, SCommitInfo *pInfo) { return tdProcessRSmaAsyncCommitImpl(pSma, pInfo); }

/**
 * @brief async commit, only applicable to Rollup SMA
 *
 * @param pSma
 * @return int32_t
 */
int32_t smaPostCommit(SSma *pSma) { return tdProcessRSmaAsyncPostCommitImpl(pSma); }

/**
 * @brief prepare rsma1/2, and set rsma trigger stat active
 *
 * @param pSma
 * @return int32_t
 */
int32_t smaBegin(SSma *pSma) {
  int32_t code = 0;
  SVnode *pVnode = pSma->pVnode;

  if ((code = tsdbBegin(VND_RSMA1(pVnode))) < 0) {
    smaError("vgId:%d, failed to begin rsma1 since %s", TD_VID(pVnode), tstrerror(code));
    goto _exit;
  }

  if ((code = tsdbBegin(VND_RSMA2(pVnode))) < 0) {
    smaError("vgId:%d, failed to begin rsma2 since %s", TD_VID(pVnode), tstrerror(code));
    goto _exit;
  }

  // set trigger stat
  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
  if (!pSmaEnv) {
    goto _exit;
  }
  SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pSmaEnv);
  int8_t     rsmaTriggerStat =
      atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_PAUSED, TASK_TRIGGER_STAT_ACTIVE);
  switch (rsmaTriggerStat) {
    case TASK_TRIGGER_STAT_PAUSED: {
      smaDebug("vgId:%d, rsma trigger stat from paused to active", TD_VID(pVnode));
      break;
    }
    case TASK_TRIGGER_STAT_INIT: {
      atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_ACTIVE);
      smaDebug("vgId:%d, rsma trigger stat from init to active", TD_VID(pVnode));
      break;
    }
    default: {
      atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_ACTIVE);
      smaWarn("vgId:%d, rsma trigger stat %" PRIi8 " is unexpected", TD_VID(pVnode), rsmaTriggerStat);
      break;
    }
  }
_exit:
  TAOS_RETURN(code);
}

extern int32_t tsdbCommitCommit(STsdb *tsdb);
int32_t        smaFinishCommit(SSma *pSma) {
  int32_t code = 0;
  int32_t lino = 0;
  SVnode *pVnode = pSma->pVnode;

  if (VND_RSMA1(pVnode) && (code = tsdbCommitCommit(VND_RSMA1(pVnode))) < 0) {
    TSDB_CHECK_CODE(code, lino, _exit);
  }
  if (VND_RSMA2(pVnode) && (code = tsdbCommitCommit(VND_RSMA2(pVnode))) < 0) {
    TSDB_CHECK_CODE(code, lino, _exit);
  }
_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  }
  TAOS_RETURN(code);
}

/**
 * @brief Rsma async commit implementation(only do some necessary light weighted task)
 *  1) set rsma stat TASK_TRIGGER_STAT_PAUSED
 *  2) Wait all running fetch task finish to fetch and put submitMsg into level 2/3 wQueue(blocking level 1 write)
 *
 * @param pSma
 * @param isCommit
 * @return int32_t
 */
extern int32_t tsdbPreCommit(STsdb *tsdb);
static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
  int32_t code = 0;
  int32_t lino = 0;

  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
  if (!pEnv) {
    return code;
  }

  SSmaStat  *pStat = SMA_ENV_STAT(pEnv);
  SRSmaStat *pRSmaStat = SMA_STAT_RSMA(pStat);
  int32_t    nLoops = 0;

  // step 1: set rsma stat
  atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_PAUSED);
  if (isCommit) {
    while (atomic_val_compare_exchange_8(RSMA_COMMIT_STAT(pRSmaStat), 0, 1) != 0) {
      TD_SMA_LOOPS_CHECK(nLoops, 1000)
    }
  }
  // step 2: wait for all triggered fetch tasks to finish
  nLoops = 0;
  while (1) {
    if (atomic_load_32(&pRSmaStat->nFetchAll) <= 0) {
      smaDebug("vgId:%d, rsma commit, type:%d, fetch tasks are all finished", SMA_VID(pSma), isCommit);
      break;
    } else {
      smaDebug("vgId:%d, rsma commit, type:%d, fetch tasks are not all finished yet", SMA_VID(pSma), isCommit);
    }
    TD_SMA_LOOPS_CHECK(nLoops, 1000);
  }

  /**
   * @brief step 3: commit should wait for all SubmitReq in buffer be consumed
   *  1) This is high cost task and should not put in asyncPreCommit originally.
   *  2) But, if put in asyncCommit, would trigger taskInfo cloning frequently.
   */
  smaInfo("vgId:%d, rsma commit, type:%d, wait for all items to be consumed, TID:%p", SMA_VID(pSma), isCommit,
          (void *)taosGetSelfPthreadId());
  nLoops = 0;
  while (atomic_load_64(&pRSmaStat->nBufItems) > 0) {
    TD_SMA_LOOPS_CHECK(nLoops, 1000);
  }
  smaInfo("vgId:%d, rsma commit, all items are consumed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());

  if (!isCommit) goto _exit;

  code = atomic_load_32(&pRSmaStat->execStat);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat));
  TSDB_CHECK_CODE(code, lino, _exit);

  smaInfo("vgId:%d, rsma commit, operator state committed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());

  // all rsma results are written completely
  STsdb *pTsdb = NULL;
  if ((pTsdb = VND_RSMA1(pSma->pVnode))) {
    code = tsdbPreCommit(pTsdb);
    TSDB_CHECK_CODE(code, lino, _exit);
  }
  if ((pTsdb = VND_RSMA2(pSma->pVnode))) {
    code = tsdbPreCommit(pTsdb);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s(%d)", SMA_VID(pSma), __func__, lino, tstrerror(code), isCommit);
  }
  TAOS_RETURN(code);
}

/**
 * @brief commit for rollup sma
 *
 * @param pSma
 * @return int32_t
 */
extern int32_t tsdbCommitBegin(STsdb *tsdb, SCommitInfo *info);
static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo) {
  int32_t code = 0;
  int32_t lino = 0;
  SVnode *pVnode = pSma->pVnode;

  if (!SMA_RSMA_ENV(pSma)) goto _exit;

  code = tsdbCommitBegin(VND_RSMA1(pVnode), pInfo);
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tsdbCommitBegin(VND_RSMA2(pVnode), pInfo);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  }
  TAOS_RETURN(code);
}

/**
 * @brief Migrate rsmaInfo from iRsmaInfo to rsmaInfo if rsma infoHash not empty.
 *
 * @param pSma
 * @return int32_t
 */
static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {
  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
  if (!pEnv) {
    TAOS_RETURN(TSDB_CODE_SUCCESS);
  }

  SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);

  // step 1: merge qTaskInfo and iQTaskInfo
  // lock
  if (1 == atomic_val_compare_exchange_8(&pRSmaStat->delFlag, 1, 0)) {
    taosWLockLatch(SMA_ENV_LOCK(pEnv));

    void *pIter = NULL;
    while ((pIter = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), pIter))) {
      tb_uid_t  *pSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
      SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)pIter;
      if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
        int32_t refVal = T_REF_VAL_GET(pRSmaInfo);
        if (refVal == 0) {
          if(taosHashRemove(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(*pSuid)) < 0) {
            smaError("vgId:%d, rsma async post commit, failed to remove rsma info for table:%" PRIi64, SMA_VID(pSma), *pSuid);
          }
        } else {
          smaDebug(
              "vgId:%d, rsma async post commit, not free rsma info since ref is %d although already deleted for "
              "table:%" PRIi64,
              SMA_VID(pSma), refVal, *pSuid);
        }

        continue;
      }
    }

    // unlock
    taosWUnLockLatch(SMA_ENV_LOCK(pEnv));
  }

  atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 0);

  TAOS_RETURN(TSDB_CODE_SUCCESS);
}

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 "sma.h"
17
18	extern SSmaMgmt smaMgmt;
19
20	static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit);
21	static int32_t tdProcessRSmaAsyncCommitImpl(SSma pSma, SCommitInfo pInfo);
22	static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma);
23	static int32_t tdUpdateQTaskInfoFiles(SSma pSma, SRSmaStat pRSmaStat);
24
25	/**
26	* @brief only applicable to Rollup SMA
27	*
28	* @param pSma
29	* @return int32_t
30	*/
31	int32_t smaPreClose(SSma *pSma) { return tdProcessRSmaAsyncPreCommitImpl(pSma, false); }	277✔
32
33	/**
34	* @brief async commit, only applicable to Rollup SMA
35	*
36	* @param pSma
37	* @return int32_t
38	*/
39	int32_t smaPrepareAsyncCommit(SSma *pSma) { return tdProcessRSmaAsyncPreCommitImpl(pSma, true); }	710✔
40
41	/**
42	* @brief async commit, only applicable to Rollup SMA
43	*
44	* @param pSma
45	* @return int32_t
46	*/
47	int32_t smaCommit(SSma pSma, SCommitInfo pInfo) { return tdProcessRSmaAsyncCommitImpl(pSma, pInfo); }	×
48
49	/**
50	* @brief async commit, only applicable to Rollup SMA
51	*
52	* @param pSma
53	* @return int32_t
54	*/
55	int32_t smaPostCommit(SSma *pSma) { return tdProcessRSmaAsyncPostCommitImpl(pSma); }	718✔
56
57	/**
58	* @brief prepare rsma1/2, and set rsma trigger stat active
59	*
60	* @param pSma
61	* @return int32_t
62	*/
63	int32_t smaBegin(SSma *pSma) {	×
64	int32_t code = 0;	×
65	SVnode *pVnode = pSma->pVnode;	×
66
67	if ((code = tsdbBegin(VND_RSMA1(pVnode))) < 0) {	×
68	smaError("vgId:%d, failed to begin rsma1 since %s", TD_VID(pVnode), tstrerror(code));	×
69	goto _exit;	×
70	}
71
72	if ((code = tsdbBegin(VND_RSMA2(pVnode))) < 0) {	×
73	smaError("vgId:%d, failed to begin rsma2 since %s", TD_VID(pVnode), tstrerror(code));	×
74	goto _exit;	×
75	}
76
77	// set trigger stat
78	SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);	×
79	if (!pSmaEnv) {	×
80	goto _exit;	×
81	}
82	SRSmaStat pRSmaStat = (SRSmaStat )SMA_ENV_STAT(pSmaEnv);	×
83	int8_t rsmaTriggerStat =
84	atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_PAUSED, TASK_TRIGGER_STAT_ACTIVE);	×
85	switch (rsmaTriggerStat) {	×
86	case TASK_TRIGGER_STAT_PAUSED: {	×
87	smaDebug("vgId:%d, rsma trigger stat from paused to active", TD_VID(pVnode));	×
88	break;	×
89	}
90	case TASK_TRIGGER_STAT_INIT: {	×
91	atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_ACTIVE);	×
92	smaDebug("vgId:%d, rsma trigger stat from init to active", TD_VID(pVnode));	×
93	break;	×
94	}
95	default: {	×
96	atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_ACTIVE);	×
97	smaWarn("vgId:%d, rsma trigger stat %" PRIi8 " is unexpected", TD_VID(pVnode), rsmaTriggerStat);	×
98	break;	×
99	}
100	}
101	_exit:	×
102	TAOS_RETURN(code);	×
103	}
104
105	extern int32_t tsdbCommitCommit(STsdb *tsdb);
106	int32_t smaFinishCommit(SSma *pSma) {	×
107	int32_t code = 0;	×
108	int32_t lino = 0;	×
109	SVnode *pVnode = pSma->pVnode;	×
110
111	if (VND_RSMA1(pVnode) && (code = tsdbCommitCommit(VND_RSMA1(pVnode))) < 0) {	×
112	TSDB_CHECK_CODE(code, lino, _exit);	×
113	}
114	if (VND_RSMA2(pVnode) && (code = tsdbCommitCommit(VND_RSMA2(pVnode))) < 0) {	×
115	TSDB_CHECK_CODE(code, lino, _exit);	×
116	}
117	_exit:	×
118	if (code) {	×
119	smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));	×
120	}
121	TAOS_RETURN(code);	×
122	}
123
124	/**
125	* @brief Rsma async commit implementation(only do some necessary light weighted task)
126	* 1) set rsma stat TASK_TRIGGER_STAT_PAUSED
127	* 2) Wait all running fetch task finish to fetch and put submitMsg into level 2/3 wQueue(blocking level 1 write)
128	*
129	* @param pSma
130	* @param isCommit
131	* @return int32_t
132	*/
133	extern int32_t tsdbPreCommit(STsdb *tsdb);
134	static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {	995✔
135	int32_t code = 0;	995✔
136	int32_t lino = 0;	995✔
137
138	SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);	995✔
139	if (!pEnv) {	995!
140	return code;	995✔
141	}
142
143	SSmaStat *pStat = SMA_ENV_STAT(pEnv);	×
144	SRSmaStat *pRSmaStat = SMA_STAT_RSMA(pStat);	×
145	int32_t nLoops = 0;	×
146
147	// step 1: set rsma stat
148	atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_PAUSED);	×
149	if (isCommit) {	×
150	while (atomic_val_compare_exchange_8(RSMA_COMMIT_STAT(pRSmaStat), 0, 1) != 0) {	×
151	TD_SMA_LOOPS_CHECK(nLoops, 1000)	×
152	}
153	}
154	// step 2: wait for all triggered fetch tasks to finish
155	nLoops = 0;	×
156	while (1) {
157	if (atomic_load_32(&pRSmaStat->nFetchAll) <= 0) {	×
158	smaDebug("vgId:%d, rsma commit, type:%d, fetch tasks are all finished", SMA_VID(pSma), isCommit);	×
159	break;	×
160	} else {
161	smaDebug("vgId:%d, rsma commit, type:%d, fetch tasks are not all finished yet", SMA_VID(pSma), isCommit);	×
162	}
163	TD_SMA_LOOPS_CHECK(nLoops, 1000);	×
164	}
165
166	/**
167	* @brief step 3: commit should wait for all SubmitReq in buffer be consumed
168	* 1) This is high cost task and should not put in asyncPreCommit originally.
169	* 2) But, if put in asyncCommit, would trigger taskInfo cloning frequently.
170	*/
171	smaInfo("vgId:%d, rsma commit, type:%d, wait for all items to be consumed, TID:%p", SMA_VID(pSma), isCommit,	×
172	(void *)taosGetSelfPthreadId());
173	nLoops = 0;	×
174	while (atomic_load_64(&pRSmaStat->nBufItems) > 0) {	×
175	TD_SMA_LOOPS_CHECK(nLoops, 1000);	×
176	}
177	smaInfo("vgId:%d, rsma commit, all items are consumed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());	×
178
179	if (!isCommit) goto _exit;	×
180
181	code = atomic_load_32(&pRSmaStat->execStat);	×
182	TSDB_CHECK_CODE(code, lino, _exit);	×
183
184	code = tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat));	×
185	TSDB_CHECK_CODE(code, lino, _exit);	×
186
187	smaInfo("vgId:%d, rsma commit, operator state committed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());	×
188
189	// all rsma results are written completely
190	STsdb *pTsdb = NULL;	×
191	if ((pTsdb = VND_RSMA1(pSma->pVnode))) {	×
192	code = tsdbPreCommit(pTsdb);	×
193	TSDB_CHECK_CODE(code, lino, _exit);	×
194	}
195	if ((pTsdb = VND_RSMA2(pSma->pVnode))) {	×
196	code = tsdbPreCommit(pTsdb);	×
197	TSDB_CHECK_CODE(code, lino, _exit);	×
198	}
199
200	_exit:	×
201	if (code) {	×
202	smaError("vgId:%d, %s failed at line %d since %s(%d)", SMA_VID(pSma), __func__, lino, tstrerror(code), isCommit);	×
203	}
204	TAOS_RETURN(code);	×
205	}
206
207	/**
208	* @brief commit for rollup sma
209	*
210	* @param pSma
211	* @return int32_t
212	*/
213	extern int32_t tsdbCommitBegin(STsdb tsdb, SCommitInfo info);
214	static int32_t tdProcessRSmaAsyncCommitImpl(SSma pSma, SCommitInfo pInfo) {	×
215	int32_t code = 0;	×
216	int32_t lino = 0;	×
217	SVnode *pVnode = pSma->pVnode;	×
218
219	if (!SMA_RSMA_ENV(pSma)) goto _exit;	×
220
221	code = tsdbCommitBegin(VND_RSMA1(pVnode), pInfo);	×
222	TSDB_CHECK_CODE(code, lino, _exit);	×
223
224	code = tsdbCommitBegin(VND_RSMA2(pVnode), pInfo);	×
225	TSDB_CHECK_CODE(code, lino, _exit);	×
226
227	_exit:	×
228	if (code) {	×
229	smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));	×
230	}
231	TAOS_RETURN(code);	×
232	}
233
234	/**
235	* @brief Migrate rsmaInfo from iRsmaInfo to rsmaInfo if rsma infoHash not empty.
236	*
237	* @param pSma
238	* @return int32_t
239	*/
240	static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {	718✔
241	SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);	718✔
242	if (!pEnv) {	718!
243	TAOS_RETURN(TSDB_CODE_SUCCESS);	719✔
244	}
245
246	SRSmaStat pRSmaStat = (SRSmaStat )SMA_ENV_STAT(pEnv);
247
248	// step 1: merge qTaskInfo and iQTaskInfo
249	// lock
250	if (1 == atomic_val_compare_exchange_8(&pRSmaStat->delFlag, 1, 0)) {	×
251	taosWLockLatch(SMA_ENV_LOCK(pEnv));	×
252
253	void *pIter = NULL;	×
254	while ((pIter = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), pIter))) {	×
255	tb_uid_t pSuid = (tb_uid_t )taosHashGetKey(pIter, NULL);	×
256	SRSmaInfo pRSmaInfo = (SRSmaInfo **)pIter;	×
257	if (RSMA_INFO_IS_DEL(pRSmaInfo)) {	×
258	int32_t refVal = T_REF_VAL_GET(pRSmaInfo);	×
259	if (refVal == 0) {	×
260	if(taosHashRemove(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(*pSuid)) < 0) {	×
261	smaError("vgId:%d, rsma async post commit, failed to remove rsma info for table:%" PRIi64, SMA_VID(pSma), *pSuid);	×
262	}
263	} else {
264	smaDebug(	×
265	"vgId:%d, rsma async post commit, not free rsma info since ref is %d although already deleted for "
266	"table:%" PRIi64,
267	SMA_VID(pSma), refVal, *pSuid);
268	}
269
270	continue;	×
271	}
272	}
273
274	// unlock
275	taosWUnLockLatch(SMA_ENV_LOCK(pEnv));	×
276	}
277
278	atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 0);	×
279
280	TAOS_RETURN(TSDB_CODE_SUCCESS);	×
281	}

taosdata / TDengine / #3819

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