#4911

Committed 04 Jan 2026 09:05AM UTC coverage: 65.028% (-0.8%) from 65.864%

Build # #4911

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch #34156

Run Details

1206 of 4524 new or added lines in 22 files covered. (26.66%)

1517 existing lines in 134 files now uncovered.

195276 of 300296 relevant lines covered (65.03%)

116931714.52 hits per line

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

65.53

/source/client/src/clientSession.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 "clientSession.h"
#include "clientInt.h"
#include "clientLog.h"

static SSessionMgt sessMgt = {0};

#define HANDLE_SESSION_CONTROL() \
  do {                           \
    if (sessMgt.inited == 0) {   \
      return TSDB_CODE_SUCCESS;  \
    }                            \
  } while (0)

static int32_t tscCheckConnStatus(STscObj *pTsc);

static int32_t sessPerUserCheckFn(int64_t *value, int64_t *limit) {
  int32_t code = 0;
  int64_t cValue = atomic_load_64(value);
  int64_t cLimit = atomic_load_64(limit);
  if (cLimit <= 0) {
    return 0;
  }

  if (cValue > cLimit) {
    code = TSDB_CODE_TSC_SESS_PER_USER_LIMIT;
  }

  return code;
}

static int32_t sessPerUserUpdateFn(int64_t *value, int64_t limit) {
  int32_t code = 0;
  int64_t ref = 0;
  if (limit > 0) {
    ref = atomic_add_fetch_64(value, limit);
  } else {
    ref = atomic_sub_fetch_64(value, -limit);
  }
  tscDebug("sessPerUserUpdateFn updated value:%" PRId64 ", limit:%" PRId64, ref, limit);
  return code;
}

static int32_t sessConnTimeCheckFn(int64_t *value, int64_t *limit) {
  int32_t code = 0;

  int64_t cValue = atomic_load_64(value);
  int64_t cLimit = atomic_load_64(limit);
  if (cLimit <= 0) {
    return code;
  }
  int64_t currentTime = taosGetTimestampMs();
  if ((cValue + cLimit * 1000) < currentTime) {
    code = TSDB_CODE_TSC_SESS_CONN_TIMEOUT;
  }

  return code;
}

static int32_t sessConnTimeUpdateFn(int64_t *value, int64_t limit) {
  int32_t code = 0;
  int64_t now = taosGetTimestampMs();
  atomic_store_64(value, now);
  return code;
}

static int32_t sessConnIdleTimeCheckFn(int64_t *value, int64_t *limit) {
  int32_t code = 0;

  int64_t currentTime = taosGetTimestampMs();
  int64_t cValue = atomic_load_64(value);
  int64_t cLimit = atomic_load_64(limit);
  if (cLimit <= 0) {
    return 0;
  }

  if ((cValue + cLimit * 1000) < currentTime) {
    code = TSDB_CODE_TSC_SESS_CONN_IDLE_TIMEOUT;
  }
  return code;
}

static int32_t sessConnIdleTimeUpdateFn(int64_t *value, int64_t limit) {
  int32_t code = 0;
  int64_t now = taosGetTimestampMs();
  atomic_store_64(value, now);

  return code;
}

static int32_t sessMaxConnCurrencyCheckFn(int64_t *value, int64_t *limit) {
  int32_t code = 0;
  int64_t cValue = atomic_load_64(value);
  int64_t cLimit = atomic_load_64(limit);
  if (cLimit <= 0) {
    return code;
  }
  if (cValue > cLimit) {
    code = TSDB_CODE_TSC_SESS_MAX_CONCURRENCY_LIMIT;
  }
  return code;
}

static int32_t sessMaxConnCurrencyUpdateFn(int64_t *value, int64_t delta) {
  int32_t code = 0;
  int64_t ref = 0;
  if (delta > 0) {
    ref = atomic_fetch_add_64(value, delta);
  } else {
    ref = atomic_fetch_sub_64(value, -delta);
  }

  tscDebug("sessMaxConnCurrencyUpdateFn updated value:%" PRId64 ", delta:%" PRId64, ref, delta);
  return code;
}

static int32_t sessVnodeCallCheckFn(int64_t *value, int64_t *limit) {
  int32_t code = 0;
  int64_t cValue = atomic_load_64(value);
  int64_t cLimit = atomic_load_64(limit);
  if (cLimit <= 0) {
    return code;
  }

  if (cValue > cLimit) {
    code = TSDB_CODE_TSC_SESS_MAX_CALL_VNODE_LIMIT;
  }
  return code;
}

static int32_t sessVnodeCallNumUpdateFn(int64_t *value, int64_t delta) {
  int32_t code = 0;
  int64_t ref = 0;
  if (delta > 0) {
    ref = atomic_fetch_add_64(value, delta);
  } else {
    ref = atomic_fetch_sub_64(value, -delta);
  }
  tscDebug("sessVnodeCallNumUpdateFn updated value:%" PRId64 ", delta:%" PRId64, ref, delta);
  return code;
}
static int32_t sessSetValueLimitFn(int64_t *pLimit, int64_t src) {
  int32_t code = 0;
  atomic_store_64(pLimit, src);
  tscDebug("sessSetValueLimitFn set limit value:%" PRId64, src);
  return code;
}

static SSessionError sessFnSet[] = {
    {SESSION_PER_USER, sessPerUserCheckFn, sessPerUserUpdateFn, sessSetValueLimitFn},
    {SESSION_CONN_TIME, sessConnTimeCheckFn, sessConnTimeUpdateFn, sessSetValueLimitFn},
    {SESSION_CONN_IDLE_TIME, sessConnIdleTimeCheckFn, sessConnIdleTimeUpdateFn, sessSetValueLimitFn},
    {SESSION_MAX_CONCURRENCY, sessMaxConnCurrencyCheckFn, sessMaxConnCurrencyUpdateFn, sessSetValueLimitFn},
    {SESSION_MAX_CALL_VNODE_NUM, sessVnodeCallCheckFn, sessVnodeCallNumUpdateFn, sessSetValueLimitFn},
};

int32_t sessMetricCreate(const char *user, SSessMetric **ppMetric) {
  HANDLE_SESSION_CONTROL();
  int32_t      code = 0;
  SSessMetric *pMetric = (SSessMetric *)taosMemoryCalloc(1, sizeof(SSessMetric));
  if (pMetric == NULL) {
    code = terrno;
    return code;
  }

  memset(pMetric->value, 0, sizeof(pMetric->value));

  pMetric->limit[SESSION_PER_USER] = sessionPerUser;
  pMetric->limit[SESSION_CONN_TIME] = sessionConnTime;
  pMetric->limit[SESSION_CONN_IDLE_TIME] = sessionConnIdleTime;
  pMetric->limit[SESSION_MAX_CONCURRENCY] = sessionMaxConcurrency;
  pMetric->limit[SESSION_MAX_CALL_VNODE_NUM] = sessionMaxCallVnodeNum;

  tstrncpy(pMetric->user, user, sizeof(pMetric->user));
  *ppMetric = pMetric;
  return code;
}

int32_t sessMetricUpdateLimit(SSessMetric *pMetric, ESessionType type, int32_t value) {
  int32_t code = 0;

  code = sessFnSet[type].limitFn(&pMetric->limit[type], value);
  return code;
}

int32_t sessMetricCheckImpl(SSessMetric *pMetric) {
  int32_t code = 0;

  for (int32_t i = 0; i < sizeof(pMetric->limit) / sizeof(pMetric->limit[0]); i++) {
    code = sessFnSet[i].checkFn(&pMetric->value[i], &pMetric->limit[i]);
    if (code != 0) {
      break;
    }
  }

  return code;
}
int32_t sessMetricCheckByTypeImpl(SSessMetric *pMetric, ESessionType type) {
  return sessFnSet[type].checkFn(&pMetric->value[type], &pMetric->limit[type]);
}

int32_t sessMetricUpdate(SSessMetric *pMetric, SSessParam *p) {
  int32_t code = 0;
  int32_t lino = 0;

  if (p->noCheck == 0) {
    code = sessMetricCheckByTypeImpl(pMetric, p->type);
    TAOS_CHECK_GOTO(code, &lino, _error);
  }

  code = sessFnSet[p->type].updateFn(&pMetric->value[p->type], p->value);
_error:
  return code;
}

int32_t sessMetricCheckValue(SSessMetric *pMetric, ESessionType type, int64_t value) {
  int32_t code = 0;
  code = sessFnSet[type].checkFn(&value, &pMetric->limit[type]);
  return code;
}
void    sessMetricDestroy(SSessMetric *pMetric) { taosMemoryFree(pMetric); }
void    sessMetricRef(SSessMetric *pMetric) { TAOS_UNUSED(atomic_add_fetch_32(&pMetric->refCnt, 1)); }
int32_t sessMetricUnref(SSessMetric *pMetric) {
  int32_t ref = atomic_sub_fetch_32(&pMetric->refCnt, 1);
  if (ref == 0) {
    tscDebug("sessMetricUnref, destroy sess metric for user:%s", pMetric->user);
  }
  return ref;
}

int32_t sessMgtInit() {
  int32_t code = 0;
  int32_t lino = 0;

  sessMgt.pSessMetricMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
  if (sessMgt.pSessMetricMap == NULL) {
    code = terrno;
    TAOS_CHECK_GOTO(code, &lino, _error);
  }
  sessMgt.inited = 1;

_error:
  if (code != 0) {
    tscError("failed to init session mgt, line:%d, code:%d", lino, code);
  }
  return code;
}

int32_t sessMgtGetOrCreateUserMetric(char *user, SSessMetric **pMetric) {
  HANDLE_SESSION_CONTROL();
  int32_t lino = 0;
  int32_t code = 0;

  SSessMetric *p = NULL;

  (void)taosThreadRwlockWrlock(&sessMgt.lock);
  SSessMetric **ppMetric = taosHashGet(sessMgt.pSessMetricMap, user, strlen(user));
  if (ppMetric == NULL || *ppMetric == NULL) {
    code = sessMetricCreate(user, &p);
    TAOS_CHECK_GOTO(code, &lino, _error);

    code = taosHashPut(sessMgt.pSessMetricMap, user, strlen(user), &p, sizeof(SSessMetric *));
    TAOS_CHECK_GOTO(code, &lino, _error);
  } else {
    p = *ppMetric;
  }

  *pMetric = p;

_error:
  (void)taosThreadRwlockUnlock(&sessMgt.lock);
  return code;
}

int32_t sessMgtUpdataLimit(char *user, ESessionType type, int32_t value) {
  HANDLE_SESSION_CONTROL();
  int32_t      code = 0;
  int32_t      lino = 0;
  SSessionMgt *pMgt = &sessMgt;

  if (type >= SESSION_MAX_TYPE || type < SESSION_PER_USER) {
    return TSDB_CODE_INVALID_PARA;
  }

  SSessMetric *pMetric = NULL;
  (void)taosThreadRwlockWrlock(&pMgt->lock);

  SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));
  if (ppMetric == NULL || *ppMetric == NULL) {
    code = sessMetricCreate(user, &pMetric);
    TAOS_CHECK_GOTO(code, &lino, _error);

    code = taosHashPut(pMgt->pSessMetricMap, user, strlen(user), &pMetric, sizeof(SSessMetric *));
    TAOS_CHECK_GOTO(code, &lino, _error);

  } else {
    pMetric = *ppMetric;
  }

  code = sessMetricUpdateLimit(pMetric, type, value);
  TAOS_CHECK_GOTO(code, &lino, _error);

_error:
  if (code != 0) {
    uError("failed to update session mgt type:%d, line:%d, code:%d", type, lino, code);
  }

  TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));

  return code;
}

int32_t sessMgtUpdateUserMetric(char *user, SSessParam *pPara) {
  HANDLE_SESSION_CONTROL();

  int32_t code = 0;
  int32_t lino = 0;

  SSessionMgt *pMgt = &sessMgt;

  SSessMetric *pMetric = NULL;
  (void)taosThreadRwlockWrlock(&pMgt->lock);

  SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));
  if (ppMetric == NULL || *ppMetric == NULL) {
    code = sessMetricCreate(user, &pMetric);
    TAOS_CHECK_GOTO(code, &lino, _error);

    code = taosHashPut(pMgt->pSessMetricMap, user, strlen(user), &pMetric, sizeof(SSessMetric *));
    TAOS_CHECK_GOTO(code, &lino, _error);
  } else {
    pMetric = *ppMetric;
  }

  code = sessMetricUpdate(pMetric, pPara);
  TAOS_CHECK_GOTO(code, &lino, _error);

_error:
  if (code != 0) {
    uError("failed to update user session metric, line:%d, code:%d", lino, code);
  }

  TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));
  return code;
}

int32_t sessMgtRemoveUser(char *user) {
  HANDLE_SESSION_CONTROL();

  int32_t      code = 0;
  int32_t      lino = 0;
  SSessionMgt *pMgt = &sessMgt;

  code = taosThreadRwlockWrlock(&pMgt->lock);
  TAOS_CHECK_GOTO(code, &lino, _error);

  SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));
  if (ppMetric != NULL && *ppMetric != NULL) {
    if (*ppMetric != NULL) {
      SSessMetric *p = *ppMetric;
      TAOS_UNUSED(taosHashRemove(pMgt->pSessMetricMap, user, strlen(user)));
      sessMetricDestroy(p);
    }
  }
_error:
  TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));
  return code;
}

void sessMgtDestroy() {
  SSessionMgt *pMgt = &sessMgt;
  int32_t      code = 0;

  if (pMgt->pSessMetricMap == NULL) {
    return;
  }

  void *p = taosHashIterate(pMgt->pSessMetricMap, NULL);
  while (p) {
    SSessMetric *pMetric = *(SSessMetric **)p;
    sessMetricDestroy(pMetric);
    p = taosHashIterate(pMgt->pSessMetricMap, p);
  }

  code = taosThreadRwlockDestroy(&pMgt->lock);
  if (code != 0) {
    uError("failed to destroy session mgt, code:%d", code);
  }
  taosHashCleanup(pMgt->pSessMetricMap);

  pMgt->pSessMetricMap = NULL;
}
int32_t tscCheckConnStatus(STscObj *pTsc) {
  HANDLE_SESSION_CONTROL();

  int32_t code = 0;
  int32_t lino = 0;

  SConnAccessInfo *p = &pTsc->sessInfo;
  SSessMetric     *pMetric = (SSessMetric *)pTsc->pSessMetric;

  code = sessMetricCheckValue(pMetric, SESSION_CONN_TIME, p->startTime);
  TAOS_CHECK_GOTO(code, &lino, _error);

  code = sessMetricCheckValue(pMetric, SESSION_CONN_IDLE_TIME, p->lastAccessTime);
  TAOS_CHECK_GOTO(code, &lino, _error);

_error:
  if (code != 0) {
    uError("failed to check connection status line:%d, code:%d", lino, code);
  }
  return code;
}

int32_t connCheckAndUpateMetric(int64_t connId) {
  HANDLE_SESSION_CONTROL();

  int32_t code = 0;
  int32_t lino = 0;

  STscObj *pTscObj = acquireTscObj(connId);
  if (pTscObj == NULL) {
    code = TSDB_CODE_INVALID_PARA;
    return code;
  }
  if (pTscObj->pSessMetric == NULL) {
    code = sessMgtGetOrCreateUserMetric((char *)pTscObj->user, (SSessMetric **)&pTscObj->pSessMetric);
    TAOS_CHECK_GOTO(code, &lino, _error);

    sessMetricRef((SSessMetric *)pTscObj->pSessMetric);
  }

  code = tscCheckConnStatus(pTscObj);
  TAOS_CHECK_GOTO(code, &lino, _error);

  updateConnAccessInfo(&pTscObj->sessInfo);

  code =
      sessMetricUpdate((SSessMetric *)pTscObj->pSessMetric, &(SSessParam){.type = SESSION_MAX_CONCURRENCY, .value = 1});
  TAOS_CHECK_GOTO(code, &lino, _error);

_error:
  if (code != 0) {
    tscError("conn:0x%" PRIx64 ", check and update metric failed at line:%d, code:%s", connId, lino, tstrerror(code));
  }

  releaseTscObj(connId);
  return code;
}

int32_t tscUpdateSessMetric(STscObj *pTscObj, SSessParam *pParam) {
  HANDLE_SESSION_CONTROL();

  int32_t code = 0;
  int32_t lino = 0;

  if (pTscObj == NULL) {
    code = TSDB_CODE_INVALID_PARA;
    return code;
  }

  SSessMetric *pMetric = (SSessMetric *)pTscObj->pSessMetric;
  if (pTscObj->pSessMetric == NULL) {
    return code;
  }

  code = sessMetricUpdate(pMetric, pParam);

  TAOS_CHECK_GOTO(code, &lino, _error);

_error:

  return code;
}

int32_t tscCheckConnSessionMetric(STscObj *pTscObj) {
  HANDLE_SESSION_CONTROL();

  int32_t code = 0;
  if (pTscObj == NULL) {
    code = TSDB_CODE_INVALID_PARA;
    return code;
  }
  code = tscCheckConnStatus(pTscObj);

_error:
  if (code != 0) {
    uError("failed to check connection session metric, code:%d", code);
  }
  return code;
}

int32_t tscRefSessMetric(STscObj *pTscObj) {
  HANDLE_SESSION_CONTROL();

  int32_t code = 0;
  int32_t lino = 0;

  SSessMetric *pMetric = NULL;
  if (pTscObj->pSessMetric != NULL) {
    return TSDB_CODE_SUCCESS;
  }
  code = sessMgtGetOrCreateUserMetric((char *)pTscObj->user, &pMetric);
  TAOS_CHECK_GOTO(code, &lino, _error);

  pTscObj->pSessMetric = pMetric;
_error:
  return code;
}
int32_t tscUnrefSessMetric(STscObj *pTscObj) {
  HANDLE_SESSION_CONTROL();
  int32_t code = 0;

  int32_t      ref = 0;
  SSessMetric *pMetric = (SSessMetric *)pTscObj->pSessMetric;
  if (pMetric != NULL) {
    ref = sessMetricUnref(pMetric);
    tscDebug("tscUnrefSessMetric conn:0x%" PRIx64 ", sess metric ref:%d", pTscObj->id, ref);
    pTscObj->pSessMetric = NULL;
  }
  return code;
}

1
2	/*
3	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
4	*
5	* This program is free software: you can use, redistribute, and/or modify
6	* it under the terms of the GNU Affero General Public License, version 3
7	* or later ("AGPL"), as published by the Free Software Foundation.
8	*
9	* This program is distributed in the hope that it will be useful, but WITHOUT
10	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11	* FITNESS FOR A PARTICULAR PURPOSE.
12	*
13	* You should have received a copy of the GNU Affero General Public License
14	* along with this program. If not, see <http://www.gnu.org/licenses/>.
15	*/
16
17	#include "clientSession.h"
18	#include "clientInt.h"
19	#include "clientLog.h"
20
21	static SSessionMgt sessMgt = {0};
22
23	#define HANDLE_SESSION_CONTROL() \
24	do { \
25	if (sessMgt.inited == 0) { \
26	return TSDB_CODE_SUCCESS; \
27	} \
28	} while (0)
29
30	static int32_t tscCheckConnStatus(STscObj *pTsc);
31
32	static int32_t sessPerUserCheckFn(int64_t value, int64_t limit) {	2,024,419✔
33	int32_t code = 0;	2,024,419✔
34	int64_t cValue = atomic_load_64(value);	2,024,419✔
35	int64_t cLimit = atomic_load_64(limit);	2,024,419✔
36	if (cLimit <= 0) {	2,024,419✔
37	return 0;	2,024,419✔
38	}
39
UNCOV 40	if (cValue > cLimit) {	×
41	code = TSDB_CODE_TSC_SESS_PER_USER_LIMIT;	×
42	}
43
UNCOV 44	return code;	×
45	}
46
47	static int32_t sessPerUserUpdateFn(int64_t *value, int64_t limit) {	4,045,408✔
48	int32_t code = 0;	4,045,408✔
49	int64_t ref = 0;	4,045,408✔
50	if (limit > 0) {	4,045,408✔
51	ref = atomic_add_fetch_64(value, limit);	2,024,429✔
52	} else {
53	ref = atomic_sub_fetch_64(value, -limit);	2,020,979✔
54	}
55	tscDebug("sessPerUserUpdateFn updated value:%" PRId64 ", limit:%" PRId64, ref, limit);	4,045,424✔
56	return code;	4,045,456✔
57	}
58
59	static int32_t sessConnTimeCheckFn(int64_t value, int64_t limit) {	742,422,650✔
60	int32_t code = 0;	742,422,650✔
61
62	int64_t cValue = atomic_load_64(value);	742,422,650✔
63	int64_t cLimit = atomic_load_64(limit);	742,420,013✔
64	if (cLimit <= 0) {	742,420,609✔
65	return code;	742,420,609✔
66	}
UNCOV 67	int64_t currentTime = taosGetTimestampMs();	×
UNCOV 68	if ((cValue + cLimit * 1000) < currentTime) {	×
69	code = TSDB_CODE_TSC_SESS_CONN_TIMEOUT;	×
70	}
71
UNCOV 72	return code;	×
73	}
74
75	static int32_t sessConnTimeUpdateFn(int64_t *value, int64_t limit) {	×
76	int32_t code = 0;	×
77	int64_t now = taosGetTimestampMs();	×
78	atomic_store_64(value, now);	×
79	return code;	×
80	}
81
82	static int32_t sessConnIdleTimeCheckFn(int64_t value, int64_t limit) {	742,421,388✔
83	int32_t code = 0;	742,421,388✔
84
85	int64_t currentTime = taosGetTimestampMs();	742,425,489✔
86	int64_t cValue = atomic_load_64(value);	742,425,489✔
87	int64_t cLimit = atomic_load_64(limit);	742,421,829✔
88	if (cLimit <= 0) {	742,423,081✔
89	return 0;	742,423,081✔
90	}
91
UNCOV 92	if ((cValue + cLimit * 1000) < currentTime) {	×
93	code = TSDB_CODE_TSC_SESS_CONN_IDLE_TIMEOUT;	×
94	}
UNCOV 95	return code;	×
96	}
97
98	static int32_t sessConnIdleTimeUpdateFn(int64_t *value, int64_t limit) {	×
99	int32_t code = 0;	×
100	int64_t now = taosGetTimestampMs();	×
101	atomic_store_64(value, now);	×
102
103	return code;	×
104	}
105
106	static int32_t sessMaxConnCurrencyCheckFn(int64_t value, int64_t limit) {	742,422,152✔
107	int32_t code = 0;	742,422,152✔
108	int64_t cValue = atomic_load_64(value);	742,422,152✔
109	int64_t cLimit = atomic_load_64(limit);	742,421,855✔
110	if (cLimit <= 0) {	742,423,932✔
111	return code;	742,423,932✔
112	}
UNCOV 113	if (cValue > cLimit) {	×
114	code = TSDB_CODE_TSC_SESS_MAX_CONCURRENCY_LIMIT;	×
115	}
UNCOV 116	return code;	×
117	}
118
119	static int32_t sessMaxConnCurrencyUpdateFn(int64_t *value, int64_t delta) {	1,490,832,731✔
120	int32_t code = 0;	1,490,832,731✔
121	int64_t ref = 0;	1,490,832,731✔
122	if (delta > 0) {	1,490,832,731✔
123	ref = atomic_fetch_add_64(value, delta);	742,424,819✔
124	} else {
125	ref = atomic_fetch_sub_64(value, -delta);	748,407,912✔
126	}
127
128	tscDebug("sessMaxConnCurrencyUpdateFn updated value:%" PRId64 ", delta:%" PRId64, ref, delta);	1,490,841,513✔
129	return code;	1,490,841,320✔
130	}
131
132	static int32_t sessVnodeCallCheckFn(int64_t value, int64_t limit) {	675,069,517✔
133	int32_t code = 0;	675,069,517✔
134	int64_t cValue = atomic_load_64(value);	675,069,517✔
135	int64_t cLimit = atomic_load_64(limit);	675,061,359✔
136	if (cLimit <= 0) {	675,057,430✔
137	return code;	675,057,430✔
138	}
139
140	if (cValue > cLimit) {	×
141	code = TSDB_CODE_TSC_SESS_MAX_CALL_VNODE_LIMIT;	×
142	}
143	return code;	×
144	}
145
146	static int32_t sessVnodeCallNumUpdateFn(int64_t *value, int64_t delta) {	×
147	int32_t code = 0;	×
148	int64_t ref = 0;	×
149	if (delta > 0) {	×
150	ref = atomic_fetch_add_64(value, delta);	×
151	} else {
152	ref = atomic_fetch_sub_64(value, -delta);	×
153	}
154	tscDebug("sessVnodeCallNumUpdateFn updated value:%" PRId64 ", delta:%" PRId64, ref, delta);	×
155	return code;	×
156	}
157	static int32_t sessSetValueLimitFn(int64_t *pLimit, int64_t src) {	5,391,495✔
158	int32_t code = 0;	5,391,495✔
159	atomic_store_64(pLimit, src);	5,391,495✔
160	tscDebug("sessSetValueLimitFn set limit value:%" PRId64, src);	5,391,495✔
161	return code;	5,391,495✔
162	}
163
164	static SSessionError sessFnSet[] = {
165	{SESSION_PER_USER, sessPerUserCheckFn, sessPerUserUpdateFn, sessSetValueLimitFn},
166	{SESSION_CONN_TIME, sessConnTimeCheckFn, sessConnTimeUpdateFn, sessSetValueLimitFn},
167	{SESSION_CONN_IDLE_TIME, sessConnIdleTimeCheckFn, sessConnIdleTimeUpdateFn, sessSetValueLimitFn},
168	{SESSION_MAX_CONCURRENCY, sessMaxConnCurrencyCheckFn, sessMaxConnCurrencyUpdateFn, sessSetValueLimitFn},
169	{SESSION_MAX_CALL_VNODE_NUM, sessVnodeCallCheckFn, sessVnodeCallNumUpdateFn, sessSetValueLimitFn},
170	};
171
172	int32_t sessMetricCreate(const char user, SSessMetric *ppMetric) {	1,167,714✔
173	HANDLE_SESSION_CONTROL();	1,167,714✔
174	int32_t code = 0;	1,167,714✔
175	SSessMetric pMetric = (SSessMetric )taosMemoryCalloc(1, sizeof(SSessMetric));	1,167,714✔
176	if (pMetric == NULL) {	1,167,714✔
177	code = terrno;	×
178	return code;	×
179	}
180
181	memset(pMetric->value, 0, sizeof(pMetric->value));	1,167,714✔
182
183	pMetric->limit[SESSION_PER_USER] = sessionPerUser;	1,167,714✔
184	pMetric->limit[SESSION_CONN_TIME] = sessionConnTime;	1,167,714✔
185	pMetric->limit[SESSION_CONN_IDLE_TIME] = sessionConnIdleTime;	1,167,714✔
186	pMetric->limit[SESSION_MAX_CONCURRENCY] = sessionMaxConcurrency;	1,167,714✔
187	pMetric->limit[SESSION_MAX_CALL_VNODE_NUM] = sessionMaxCallVnodeNum;	1,167,714✔
188
189	tstrncpy(pMetric->user, user, sizeof(pMetric->user));	1,167,714✔
190	*ppMetric = pMetric;	1,167,714✔
191	return code;	1,167,714✔
192	}
193
194	int32_t sessMetricUpdateLimit(SSessMetric *pMetric, ESessionType type, int32_t value) {	5,391,495✔
195	int32_t code = 0;	5,391,495✔
196
197	code = sessFnSet[type].limitFn(&pMetric->limit[type], value);	5,391,495✔
198	return code;	5,391,495✔
199	}
200
201	int32_t sessMetricCheckImpl(SSessMetric *pMetric) {	×
202	int32_t code = 0;	×
203
204	for (int32_t i = 0; i < sizeof(pMetric->limit) / sizeof(pMetric->limit[0]); i++) {	×
205	code = sessFnSet[i].checkFn(&pMetric->value[i], &pMetric->limit[i]);	×
206	if (code != 0) {	×
207	break;	×
208	}
209	}
210
211	return code;	×
212	}
213	int32_t sessMetricCheckByTypeImpl(SSessMetric *pMetric, ESessionType type) {	744,446,617✔
214	return sessFnSet[type].checkFn(&pMetric->value[type], &pMetric->limit[type]);	744,446,617✔
215	}
216
217	int32_t sessMetricUpdate(SSessMetric pMetric, SSessParam p) {	1,494,880,368✔
218	int32_t code = 0;	1,494,880,368✔
219	int32_t lino = 0;	1,494,880,368✔
220
221	if (p->noCheck == 0) {	1,494,885,331✔
222	code = sessMetricCheckByTypeImpl(pMetric, p->type);	744,450,465✔
223	TAOS_CHECK_GOTO(code, &lino, _error);	744,449,955✔
224	}
225
226	code = sessFnSet[p->type].updateFn(&pMetric->value[p->type], p->value);	1,494,879,661✔
227	_error:	1,494,885,891✔
228	return code;	1,494,887,024✔
229	}
230
231	int32_t sessMetricCheckValue(SSessMetric *pMetric, ESessionType type, int64_t value) {	2,147,483,647✔
232	int32_t code = 0;	2,147,483,647✔
233	code = sessFnSet[type].checkFn(&value, &pMetric->limit[type]);	2,147,483,647✔
234	return code;	2,147,483,647✔
235	}
236	void sessMetricDestroy(SSessMetric *pMetric) { taosMemoryFree(pMetric); }	1,167,714✔
237	void sessMetricRef(SSessMetric *pMetric) { TAOS_UNUSED(atomic_add_fetch_32(&pMetric->refCnt, 1)); }	2,024,429✔
238	int32_t sessMetricUnref(SSessMetric *pMetric) {	2,020,828✔
239	int32_t ref = atomic_sub_fetch_32(&pMetric->refCnt, 1);	2,020,828✔
240	if (ref == 0) {	2,020,828✔
241	tscDebug("sessMetricUnref, destroy sess metric for user:%s", pMetric->user);	1,424,439✔
242	}
243	return ref;	2,021,027✔
244	}
245
246	int32_t sessMgtInit() {	1,155,854✔
247	int32_t code = 0;	1,155,854✔
248	int32_t lino = 0;	1,155,854✔
249
250	sessMgt.pSessMetricMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);	1,155,854✔
251	if (sessMgt.pSessMetricMap == NULL) {	1,155,854✔
252	code = terrno;	×
253	TAOS_CHECK_GOTO(code, &lino, _error);	×
254	}
255	sessMgt.inited = 1;	1,155,854✔
256
257	_error:	1,155,854✔
258	if (code != 0) {	1,155,854✔
259	tscError("failed to init session mgt, line:%d, code:%d", lino, code);	×
260	}
261	return code;	1,155,854✔
262	}
263
264	int32_t sessMgtGetOrCreateUserMetric(char user, SSessMetric *pMetric) {	2,024,374✔
265	HANDLE_SESSION_CONTROL();	2,024,374✔
266	int32_t lino = 0;	2,024,374✔
267	int32_t code = 0;	2,024,374✔
268
269	SSessMetric *p = NULL;	2,024,374✔
270
271	(void)taosThreadRwlockWrlock(&sessMgt.lock);	2,024,370✔
272	SSessMetric **ppMetric = taosHashGet(sessMgt.pSessMetricMap, user, strlen(user));	2,024,429✔
273	if (ppMetric == NULL \|\| *ppMetric == NULL) {	2,024,429✔
274	code = sessMetricCreate(user, &p);	1,167,714✔
275	TAOS_CHECK_GOTO(code, &lino, _error);	1,167,714✔
276
277	code = taosHashPut(sessMgt.pSessMetricMap, user, strlen(user), &p, sizeof(SSessMetric *));	1,167,714✔
278	TAOS_CHECK_GOTO(code, &lino, _error);	1,167,714✔
279	} else {
280	p = *ppMetric;	856,715✔
281	}
282
283	*pMetric = p;	2,024,429✔
284
285	_error:	2,024,429✔
286	(void)taosThreadRwlockUnlock(&sessMgt.lock);	2,024,429✔
287	return code;	2,024,429✔
288	}
289
290	int32_t sessMgtUpdataLimit(char *user, ESessionType type, int32_t value) {	5,391,495✔
291	HANDLE_SESSION_CONTROL();	5,391,495✔
292	int32_t code = 0;	5,391,495✔
293	int32_t lino = 0;	5,391,495✔
294	SSessionMgt *pMgt = &sessMgt;	5,391,495✔
295
296	if (type >= SESSION_MAX_TYPE \|\| type < SESSION_PER_USER) {	5,391,495✔
297	return TSDB_CODE_INVALID_PARA;	×
298	}
299
300	SSessMetric *pMetric = NULL;	5,391,495✔
301	(void)taosThreadRwlockWrlock(&pMgt->lock);	5,391,495✔
302
303	SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));	5,391,495✔
304	if (ppMetric == NULL \|\| *ppMetric == NULL) {	5,391,495✔
305	code = sessMetricCreate(user, &pMetric);	×
306	TAOS_CHECK_GOTO(code, &lino, _error);	×
307
308	code = taosHashPut(pMgt->pSessMetricMap, user, strlen(user), &pMetric, sizeof(SSessMetric *));	×
309	TAOS_CHECK_GOTO(code, &lino, _error);	×
310
311	} else {
312	pMetric = *ppMetric;	5,391,495✔
313	}
314
315	code = sessMetricUpdateLimit(pMetric, type, value);	5,391,495✔
316	TAOS_CHECK_GOTO(code, &lino, _error);	5,391,495✔
317
318	_error:	5,391,495✔
319	if (code != 0) {	5,391,495✔
320	uError("failed to update session mgt type:%d, line:%d, code:%d", type, lino, code);	×
321	}
322
323	TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));	5,391,495✔
324
325	return code;	5,391,495✔
326	}
327
328	int32_t sessMgtUpdateUserMetric(char user, SSessParam pPara) {	×
329	HANDLE_SESSION_CONTROL();	×
330
331	int32_t code = 0;	×
332	int32_t lino = 0;	×
333
334	SSessionMgt *pMgt = &sessMgt;	×
335
336	SSessMetric *pMetric = NULL;	×
337	(void)taosThreadRwlockWrlock(&pMgt->lock);	×
338
339	SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));	×
340	if (ppMetric == NULL \|\| *ppMetric == NULL) {	×
341	code = sessMetricCreate(user, &pMetric);	×
342	TAOS_CHECK_GOTO(code, &lino, _error);	×
343
344	code = taosHashPut(pMgt->pSessMetricMap, user, strlen(user), &pMetric, sizeof(SSessMetric *));	×
345	TAOS_CHECK_GOTO(code, &lino, _error);	×
346	} else {
347	pMetric = *ppMetric;	×
348	}
349
350	code = sessMetricUpdate(pMetric, pPara);	×
351	TAOS_CHECK_GOTO(code, &lino, _error);	×
352
353	_error:	×
354	if (code != 0) {	×
355	uError("failed to update user session metric, line:%d, code:%d", lino, code);	×
356	}
357
358	TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));	×
359	return code;	×
360	}
361
362	int32_t sessMgtRemoveUser(char *user) {	×
363	HANDLE_SESSION_CONTROL();	×
364
365	int32_t code = 0;	×
366	int32_t lino = 0;	×
367	SSessionMgt *pMgt = &sessMgt;	×
368
369	code = taosThreadRwlockWrlock(&pMgt->lock);	×
370	TAOS_CHECK_GOTO(code, &lino, _error);	×
371
372	SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));	×
373	if (ppMetric != NULL && *ppMetric != NULL) {	×
374	if (*ppMetric != NULL) {	×
375	SSessMetric p = ppMetric;	×
376	TAOS_UNUSED(taosHashRemove(pMgt->pSessMetricMap, user, strlen(user)));	×
377	sessMetricDestroy(p);	×
378	}
379	}
380	_error:	×
381	TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));	×
382	return code;	×
383	}
384
385	void sessMgtDestroy() {	1,155,868✔
386	SSessionMgt *pMgt = &sessMgt;	1,155,868✔
387	int32_t code = 0;	1,155,868✔
388
389	if (pMgt->pSessMetricMap == NULL) {	1,155,868✔
390	return;	14✔
391	}
392
393	void *p = taosHashIterate(pMgt->pSessMetricMap, NULL);	1,155,854✔
394	while (p) {	2,323,568✔
395	SSessMetric pMetric = (SSessMetric **)p;	1,167,714✔
396	sessMetricDestroy(pMetric);	1,167,714✔
397	p = taosHashIterate(pMgt->pSessMetricMap, p);	1,167,714✔
398	}
399
400	code = taosThreadRwlockDestroy(&pMgt->lock);	1,155,854✔
401	if (code != 0) {	1,155,854✔
402	uError("failed to destroy session mgt, code:%d", code);	×
403	}
404	taosHashCleanup(pMgt->pSessMetricMap);	1,155,854✔
405
406	pMgt->pSessMetricMap = NULL;	1,155,854✔
407	}
408	int32_t tscCheckConnStatus(STscObj *pTsc) {	742,424,727✔
409	HANDLE_SESSION_CONTROL();	742,424,727✔
410
411	int32_t code = 0;	742,424,727✔
412	int32_t lino = 0;	742,424,727✔
413
414	SConnAccessInfo *p = &pTsc->sessInfo;	742,425,964✔
415	SSessMetric pMetric = (SSessMetric )pTsc->pSessMetric;	742,424,779✔
416
417	code = sessMetricCheckValue(pMetric, SESSION_CONN_TIME, p->startTime);	742,422,901✔
418	TAOS_CHECK_GOTO(code, &lino, _error);	742,418,142✔
419
420	code = sessMetricCheckValue(pMetric, SESSION_CONN_IDLE_TIME, p->lastAccessTime);	742,418,142✔
421	TAOS_CHECK_GOTO(code, &lino, _error);	742,423,267✔
422
423	_error:	742,423,267✔
424	if (code != 0) {	742,424,071✔
425	uError("failed to check connection status line:%d, code:%d", lino, code);	×
426	}
427	return code;	742,420,977✔
428	}
429
430	int32_t connCheckAndUpateMetric(int64_t connId) {	742,422,573✔
431	HANDLE_SESSION_CONTROL();	742,422,573✔
432
433	int32_t code = 0;	742,422,573✔
434	int32_t lino = 0;	742,422,573✔
435
436	STscObj *pTscObj = acquireTscObj(connId);	742,423,979✔
437	if (pTscObj == NULL) {	742,427,556✔
438	code = TSDB_CODE_INVALID_PARA;	×
439	return code;	×
440	}
441	if (pTscObj->pSessMetric == NULL) {	742,427,556✔
442	code = sessMgtGetOrCreateUserMetric((char )pTscObj->user, (SSessMetric *)&pTscObj->pSessMetric);	×
443	TAOS_CHECK_GOTO(code, &lino, _error);	×
444
445	sessMetricRef((SSessMetric *)pTscObj->pSessMetric);	×
446	}
447
448	code = tscCheckConnStatus(pTscObj);	742,427,556✔
449	TAOS_CHECK_GOTO(code, &lino, _error);	742,420,508✔
450
451	updateConnAccessInfo(&pTscObj->sessInfo);	742,420,508✔
452
453	code =
454	sessMetricUpdate((SSessMetric *)pTscObj->pSessMetric, &(SSessParam){.type = SESSION_MAX_CONCURRENCY, .value = 1});	742,423,346✔
455	TAOS_CHECK_GOTO(code, &lino, _error);	742,420,887✔
456
457	_error:	742,420,887✔
458	if (code != 0) {	742,423,686✔
459	tscError("conn:0x%" PRIx64 ", check and update metric failed at line:%d, code:%s", connId, lino, tstrerror(code));	×
460	}
461
462	releaseTscObj(connId);	742,423,686✔
463	return code;	742,420,383✔
464	}
465
466	int32_t tscUpdateSessMetric(STscObj pTscObj, SSessParam pParam) {	752,525,058✔
467	HANDLE_SESSION_CONTROL();	752,525,058✔
468
469	int32_t code = 0;	752,525,058✔
470	int32_t lino = 0;	752,525,058✔
471
472	if (pTscObj == NULL) {	752,526,341✔
473	code = TSDB_CODE_INVALID_PARA;	×
474	return code;	×
475	}
476
477	SSessMetric pMetric = (SSessMetric )pTscObj->pSessMetric;	752,526,341✔
478	if (pTscObj->pSessMetric == NULL) {	752,528,575✔
479	return code;	67,257✔
480	}
481
482	code = sessMetricUpdate(pMetric, pParam);	752,458,851✔
483
484	TAOS_CHECK_GOTO(code, &lino, _error);	752,459,445✔
485
486	_error:	752,459,445✔
487
488	return code;	752,459,445✔
489	}
490
491	int32_t tscCheckConnSessionMetric(STscObj *pTscObj) {	×
492	HANDLE_SESSION_CONTROL();	×
493
494	int32_t code = 0;	×
495	if (pTscObj == NULL) {	×
496	code = TSDB_CODE_INVALID_PARA;	×
497	return code;	×
498	}
499	code = tscCheckConnStatus(pTscObj);	×
500
501	_error:	×
502	if (code != 0) {	×
503	uError("failed to check connection session metric, code:%d", code);	×
504	}
505	return code;	×
506	}
507
508	int32_t tscRefSessMetric(STscObj *pTscObj) {	2,024,165✔
509	HANDLE_SESSION_CONTROL();	2,024,165✔
510
511	int32_t code = 0;	2,024,165✔
512	int32_t lino = 0;	2,024,165✔
513
514	SSessMetric *pMetric = NULL;	2,024,165✔
515	if (pTscObj->pSessMetric != NULL) {	2,024,165✔
516	return TSDB_CODE_SUCCESS;	×
517	}
518	code = sessMgtGetOrCreateUserMetric((char *)pTscObj->user, &pMetric);	2,024,080✔
519	TAOS_CHECK_GOTO(code, &lino, _error);	2,024,429✔
520
521	pTscObj->pSessMetric = pMetric;	2,024,429✔
522	_error:	2,024,429✔
523	return code;	2,024,429✔
524	}
525	int32_t tscUnrefSessMetric(STscObj *pTscObj) {	2,026,875✔
526	HANDLE_SESSION_CONTROL();	2,026,875✔
527	int32_t code = 0;	2,026,875✔
528
529	int32_t ref = 0;	2,026,875✔
530	SSessMetric pMetric = (SSessMetric )pTscObj->pSessMetric;	2,026,875✔
531	if (pMetric != NULL) {	2,026,875✔
532	ref = sessMetricUnref(pMetric);	2,020,995✔
533	tscDebug("tscUnrefSessMetric conn:0x%" PRIx64 ", sess metric ref:%d", pTscObj->id, ref);	2,021,027✔
534	pTscObj->pSessMetric = NULL;	2,021,027✔
535	}
536	return code;	2,026,875✔
537	}

taosdata / TDengine / #4911

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