#4943

Committed 30 Jan 2026 06:19AM UTC coverage: 66.718% (-0.07%) from 66.788%

Build # #4943

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 #34453

Run Details

1122 of 2018 new or added lines in 72 files covered. (55.6%)

823 existing lines in 156 files now uncovered.

204811 of 306978 relevant lines covered (66.72%)

123993567.34 hits per line

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

69.88

/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,548,443✔
33	int32_t code = 0;	2,548,443✔
34	int64_t cValue = atomic_load_64(value);	2,548,443✔
35	int64_t cLimit = atomic_load_64(limit);	2,548,428✔
36	if (cLimit <= 0) {	2,548,456✔
37	return 0;	2,536,366✔
38	}
39
40	if (cValue > cLimit) {	12,090✔
41	code = TSDB_CODE_TSC_SESS_PER_USER_LIMIT;	×
42	}
43
44	return code;	12,090✔
45	}
46
47	static int32_t sessPerUserUpdateFn(int64_t *value, int64_t limit) {	5,093,315✔
48	int32_t code = 0;	5,093,315✔
49	int64_t ref = 0;	5,093,315✔
50	if (limit > 0) {	5,093,315✔
51	ref = atomic_add_fetch_64(value, limit);	2,548,456✔
52	} else {
53	ref = atomic_sub_fetch_64(value, -limit);	2,544,859✔
54	}
55	tscDebug("sessPerUserUpdateFn updated value:%" PRId64 ", limit:%" PRId64, ref, limit);	5,093,382✔
56	return code;	5,093,382✔
57	}
58
59	static int32_t sessConnTimeCheckFn(int64_t value, int64_t limit) {	868,416,947✔
60	int32_t code = 0;	868,416,947✔
61
62	int64_t cValue = atomic_load_64(value);	868,416,947✔
63	int64_t cLimit = atomic_load_64(limit);	868,417,028✔
64	if (cLimit <= 0) {	868,418,257✔
65	return code;	868,409,422✔
66	}
67	int64_t currentTime = taosGetTimestampMs();	8,835✔
68	if ((cValue + cLimit * 1000) < currentTime) {	8,835✔
69	code = TSDB_CODE_TSC_SESS_CONN_TIMEOUT;	155✔
70	}
71
72	return code;	8,835✔
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) {	868,416,419✔
83	int32_t code = 0;	868,416,419✔
84
85	int64_t currentTime = taosGetTimestampMs();	868,423,056✔
86	int64_t cValue = atomic_load_64(value);	868,423,056✔
87	int64_t cLimit = atomic_load_64(limit);	868,421,947✔
88	if (cLimit <= 0) {	868,422,021✔
89	return 0;	868,421,866✔
90	}
91
92	if ((cValue + cLimit * 1000) < currentTime) {	155✔
93	code = TSDB_CODE_TSC_SESS_CONN_IDLE_TIMEOUT;	155✔
94	}
95	return code;	155✔
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) {	868,413,105✔
107	int32_t code = 0;	868,413,105✔
108	int64_t cValue = atomic_load_64(value);	868,413,105✔
109	int64_t cLimit = atomic_load_64(limit);	868,413,681✔
110	if (cLimit <= 0) {	868,420,139✔
111	return code;	868,420,139✔
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,747,377,164✔
120	int32_t code = 0;	1,747,377,164✔
121	int64_t ref = 0;	1,747,377,164✔
122	if (delta > 0) {	1,747,377,164✔
123	ref = atomic_fetch_add_64(value, delta);	868,418,571✔
124	} else {
125	ref = atomic_fetch_sub_64(value, -delta);	878,958,593✔
126	}
127
128	tscDebug("sessMaxConnCurrencyUpdateFn updated value:%" PRId64 ", delta:%" PRId64, ref, delta);	1,747,397,977✔
129	return code;	1,747,396,798✔
130	}
131
132	static int32_t sessVnodeCallCheckFn(int64_t value, int64_t limit) {	791,256,605✔
133	int32_t code = 0;	791,256,605✔
134	int64_t cValue = atomic_load_64(value);	791,256,605✔
135	int64_t cLimit = atomic_load_64(limit);	791,258,268✔
136	if (cLimit <= 0) {	791,267,198✔
137	return code;	791,263,943✔
138	}
139
140	if (cValue > cLimit) {	3,255✔
141	code = TSDB_CODE_TSC_SESS_MAX_CALL_VNODE_LIMIT;	3,100✔
142	}
143	return code;	3,255✔
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) {	212,476,525✔
158	int32_t code = 0;	212,476,525✔
159	atomic_store_64(pLimit, src);	212,476,525✔
160	tscDebug("sessSetValueLimitFn set limit value:%" PRId64, src);	212,476,525✔
161	return code;	212,476,525✔
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,313,833✔
173	HANDLE_SESSION_CONTROL();	1,313,833✔
174	int32_t code = 0;	1,313,833✔
175	SSessMetric pMetric = (SSessMetric )taosMemoryCalloc(1, sizeof(SSessMetric));	1,313,833✔
176	if (pMetric == NULL) {	1,313,833✔
177	code = terrno;	×
178	return code;	×
179	}
180
181	memset(pMetric->value, 0, sizeof(pMetric->value));	1,313,833✔
182
183	pMetric->limit[SESSION_PER_USER] = sessionPerUser;	1,313,833✔
184	pMetric->limit[SESSION_CONN_TIME] = sessionConnTime;	1,313,833✔
185	pMetric->limit[SESSION_CONN_IDLE_TIME] = sessionConnIdleTime;	1,313,833✔
186	pMetric->limit[SESSION_MAX_CONCURRENCY] = sessionMaxConcurrency;	1,313,833✔
187	pMetric->limit[SESSION_MAX_CALL_VNODE_NUM] = sessionMaxCallVnodeNum;	1,313,833✔
188
189	tstrncpy(pMetric->user, user, sizeof(pMetric->user));	1,313,833✔
190	*ppMetric = pMetric;	1,313,833✔
191	return code;	1,313,833✔
192	}
193
194	int32_t sessMetricUpdateLimit(SSessMetric *pMetric, ESessionType type, int32_t value) {	212,476,525✔
195	int32_t code = 0;	212,476,525✔
196
197	code = sessFnSet[type].limitFn(&pMetric->limit[type], value);	212,476,525✔
198	return code;	212,476,525✔
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) {	870,962,994✔
214	return sessFnSet[type].checkFn(&pMetric->value[type], &pMetric->limit[type]);	870,962,994✔
215	}
216
217	int32_t sessMetricUpdate(SSessMetric pMetric, SSessParam p) {	1,752,467,980✔
218	int32_t code = 0;	1,752,467,980✔
219	int32_t lino = 0;	1,752,467,980✔
220
221	if (p->noCheck == 0) {	1,752,472,856✔
222	code = sessMetricCheckByTypeImpl(pMetric, p->type);	870,963,800✔
223	TAOS_CHECK_GOTO(code, &lino, _error);	870,970,880✔
224	}
225
226	code = sessFnSet[p->type].updateFn(&pMetric->value[p->type], p->value);	1,752,491,899✔
227	_error:	1,752,489,718✔
228	return code;	1,752,490,123✔
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,313,833✔
237	void sessMetricRef(SSessMetric *pMetric) { TAOS_UNUSED(atomic_add_fetch_32(&pMetric->refCnt, 1)); }	2,548,456✔
238	int32_t sessMetricUnref(SSessMetric *pMetric) {	2,544,887✔
239	int32_t ref = atomic_sub_fetch_32(&pMetric->refCnt, 1);	2,544,887✔
240	if (ref == 0) {	2,544,852✔
241	tscDebug("sessMetricUnref, destroy sess metric for user:%s", pMetric->user);	1,616,894✔
242	}
243	return ref;	2,544,852✔
244	}
245
246	int32_t sessMgtInit() {	1,297,512✔
247	int32_t code = 0;	1,297,512✔
248	int32_t lino = 0;	1,297,512✔
249
250	sessMgt.pSessMetricMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);	1,297,512✔
251	if (sessMgt.pSessMetricMap == NULL) {	1,297,512✔
252	code = terrno;	×
253	TAOS_CHECK_GOTO(code, &lino, _error);	×
254	}
255	sessMgt.inited = 1;	1,297,512✔
256
257	_error:	1,297,512✔
258	if (code != 0) {	1,297,512✔
259	tscError("failed to init session mgt, line:%d, code:%d", lino, code);	×
260	}
261	return code;	1,297,512✔
262	}
263
264	int32_t sessMgtGetOrCreateUserMetric(char user, SSessMetric *pMetric) {	2,548,418✔
265	HANDLE_SESSION_CONTROL();	2,548,418✔
266	int32_t lino = 0;	2,548,418✔
267	int32_t code = 0;	2,548,362✔
268
269	SSessMetric *p = NULL;	2,548,362✔
270
271	(void)taosThreadRwlockWrlock(&sessMgt.lock);	2,548,344✔
272	SSessMetric **ppMetric = taosHashGet(sessMgt.pSessMetricMap, user, strlen(user));	2,548,495✔
273	if (ppMetric == NULL \|\| *ppMetric == NULL) {	2,548,495✔
274	code = sessMetricCreate(user, &p);	1,313,833✔
275	TAOS_CHECK_GOTO(code, &lino, _error);	1,313,833✔
276
277	code = taosHashPut(sessMgt.pSessMetricMap, user, strlen(user), &p, sizeof(SSessMetric *));	1,313,833✔
278	TAOS_CHECK_GOTO(code, &lino, _error);	1,313,833✔
279	} else {
280	p = *ppMetric;	1,234,662✔
281	}
282
283	*pMetric = p;	2,548,495✔
284
285	_error:	2,548,495✔
286	(void)taosThreadRwlockUnlock(&sessMgt.lock);	2,548,495✔
287	return code;	2,548,495✔
288	}
289
290	int32_t sessMgtUpdataLimit(char *user, ESessionType type, int32_t value) {	212,476,525✔
291	HANDLE_SESSION_CONTROL();	212,476,525✔
292	int32_t code = 0;	212,476,525✔
293	int32_t lino = 0;	212,476,525✔
294	SSessionMgt *pMgt = &sessMgt;	212,476,525✔
295
296	if (type >= SESSION_MAX_TYPE \|\| type < SESSION_PER_USER) {	212,476,525✔
297	return TSDB_CODE_INVALID_PARA;	×
298	}
299
300	SSessMetric *pMetric = NULL;	212,476,525✔
301	(void)taosThreadRwlockWrlock(&pMgt->lock);	212,476,525✔
302
303	SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));	212,476,525✔
304	if (ppMetric == NULL \|\| *ppMetric == NULL) {	212,476,525✔
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;	212,476,525✔
313	}
314
315	code = sessMetricUpdateLimit(pMetric, type, value);	212,476,525✔
316	TAOS_CHECK_GOTO(code, &lino, _error);	212,476,525✔
317
318	_error:	212,476,525✔
319	if (code != 0) {	212,476,525✔
320	uError("failed to update session mgt type:%d, line:%d, code:%d", type, lino, code);	×
321	}
322
323	TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));	212,476,525✔
324
325	return code;	212,476,525✔
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,297,566✔
386	SSessionMgt *pMgt = &sessMgt;	1,297,566✔
387	int32_t code = 0;	1,297,566✔
388
389	if (pMgt->pSessMetricMap == NULL) {	1,297,566✔
390	return;	54✔
391	}
392
393	void *p = taosHashIterate(pMgt->pSessMetricMap, NULL);	1,297,512✔
394	while (p) {	2,611,345✔
395	SSessMetric pMetric = (SSessMetric **)p;	1,313,833✔
396	sessMetricDestroy(pMetric);	1,313,833✔
397	p = taosHashIterate(pMgt->pSessMetricMap, p);	1,313,833✔
398	}
399
400	code = taosThreadRwlockDestroy(&pMgt->lock);	1,297,512✔
401	if (code != 0) {	1,297,512✔
402	uError("failed to destroy session mgt, code:%d", code);	×
403	}
404	taosHashCleanup(pMgt->pSessMetricMap);	1,297,512✔
405
406	pMgt->pSessMetricMap = NULL;	1,297,512✔
407	}
408	int32_t tscCheckConnStatus(STscObj *pTsc) {	868,420,367✔
409	HANDLE_SESSION_CONTROL();	868,420,367✔
410
411	int32_t code = 0;	868,420,367✔
412	int32_t lino = 0;	868,420,367✔
413
414	SConnAccessInfo *p = &pTsc->sessInfo;	868,423,079✔
415	SSessMetric pMetric = (SSessMetric )pTsc->pSessMetric;	868,422,675✔
416
417	code = sessMetricCheckValue(pMetric, SESSION_CONN_TIME, p->startTime);	868,423,824✔
418	TAOS_CHECK_GOTO(code, &lino, _error);	868,420,475✔
419
420	code = sessMetricCheckValue(pMetric, SESSION_CONN_IDLE_TIME, p->lastAccessTime);	868,420,320✔
421	TAOS_CHECK_GOTO(code, &lino, _error);	868,421,963✔
422
423	_error:	868,422,118✔
424	if (code != 0) {	868,421,340✔
425	uError("failed to check connection status line:%d, code:%d", lino, code);	310✔
426	}
427	return code;	868,418,775✔
428	}
429
430	int32_t connCheckAndUpateMetric(int64_t connId) {	868,417,330✔
431	HANDLE_SESSION_CONTROL();	868,417,330✔
432
433	int32_t code = 0;	868,417,330✔
434	int32_t lino = 0;	868,417,330✔
435
436	STscObj *pTscObj = acquireTscObj(connId);	868,420,137✔
437	if (pTscObj == NULL) {	868,425,017✔
438	code = TSDB_CODE_INVALID_PARA;	×
439	return code;	×
440	}
441	if (pTscObj->pSessMetric == NULL) {	868,425,017✔
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);	868,424,605✔
449	TAOS_CHECK_GOTO(code, &lino, _error);	868,416,004✔
450
451	updateConnAccessInfo(&pTscObj->sessInfo);	868,415,694✔
452
453	code =
454	sessMetricUpdate((SSessMetric *)pTscObj->pSessMetric, &(SSessParam){.type = SESSION_MAX_CONCURRENCY, .value = 1});	868,418,685✔
455	TAOS_CHECK_GOTO(code, &lino, _error);	868,424,248✔
456
457	_error:	868,424,558✔
458	if (code != 0) {	868,424,969✔
459	tscError("conn:0x%" PRIx64 ", check and update metric failed at line:%d, code:%s", connId, lino, tstrerror(code));	310✔
460	}
461
462	releaseTscObj(connId);	868,424,969✔
463	return code;	868,415,526✔
464	}
465
466	int32_t tscUpdateSessMetric(STscObj pTscObj, SSessParam pParam) {	884,136,082✔
467	HANDLE_SESSION_CONTROL();	884,136,082✔
468
469	int32_t code = 0;	884,136,082✔
470	int32_t lino = 0;	884,136,082✔
471
472	if (pTscObj == NULL) {	884,136,492✔
473	code = TSDB_CODE_INVALID_PARA;	×
474	return code;	×
475	}
476
477	SSessMetric pMetric = (SSessMetric )pTscObj->pSessMetric;	884,136,492✔
478	if (pTscObj->pSessMetric == NULL) {	884,136,517✔
479	return code;	75,322✔
480	}
481
482	code = sessMetricUpdate(pMetric, pParam);	884,069,334✔
483
484	TAOS_CHECK_GOTO(code, &lino, _error);	884,066,987✔
485
486	_error:	884,066,987✔
487
488	return code;	884,066,987✔
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,548,356✔
509	HANDLE_SESSION_CONTROL();	2,548,356✔
510
511	int32_t code = 0;	2,548,356✔
512	int32_t lino = 0;	2,548,356✔
513
514	SSessMetric *pMetric = NULL;	2,548,356✔
515	if (pTscObj->pSessMetric != NULL) {	2,548,356✔
516	return TSDB_CODE_SUCCESS;	×
517	}
518	code = sessMgtGetOrCreateUserMetric((char *)pTscObj->user, &pMetric);	2,547,113✔
519	TAOS_CHECK_GOTO(code, &lino, _error);	2,548,495✔
520
521	pTscObj->pSessMetric = pMetric;	2,548,495✔
522	_error:	2,548,495✔
523	return code;	2,548,495✔
524	}
525	int32_t tscUnrefSessMetric(STscObj *pTscObj) {	2,558,859✔
526	HANDLE_SESSION_CONTROL();	2,558,859✔
527	int32_t code = 0;	2,558,859✔
528
529	int32_t ref = 0;	2,558,859✔
530	SSessMetric pMetric = (SSessMetric )pTscObj->pSessMetric;	2,558,859✔
531	if (pMetric != NULL) {	2,558,859✔
532	ref = sessMetricUnref(pMetric);	2,544,852✔
533	tscDebug("tscUnrefSessMetric conn:0x%" PRIx64 ", sess metric ref:%d", pTscObj->id, ref);	2,544,852✔
534	pTscObj->pSessMetric = NULL;	2,544,852✔
535	}
536	return code;	2,558,789✔
537	}

taosdata / TDengine / #4943

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