#4986

Committed 15 Mar 2026 08:32AM UTC coverage: 37.305% (-31.3%) from 68.601%

Build # #4986

Build Type

push

travis-ci

Committed by

tomchon

Commit Message

test: keep docs and unit test

Coverage Stats

125478 of 336361 relevant lines covered (37.3%)

1134847.06 hits per line

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

77.02

/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) {	10,299✔
33	int32_t code = 0;	10,299✔
34	int64_t cValue = atomic_load_64(value);	10,299✔
35	int64_t cLimit = atomic_load_64(limit);	10,299✔
36	if (cLimit <= 0) {	10,299✔
37	return 0;	265✔
38	}
39
40	if (cValue > cLimit) {	10,034✔
41	code = TSDB_CODE_TSC_SESS_PER_USER_LIMIT;	10✔
42	}
43
44	return code;	10,034✔
45	}
46
47	static int32_t sessPerUserUpdateFn(int64_t *value, int64_t limit) {	10,553✔
48	int32_t code = 0;	10,553✔
49	int64_t ref = 0;	10,553✔
50	if (limit > 0) {	10,553✔
51	ref = atomic_add_fetch_64(value, limit);	10,289✔
52	} else {
53	ref = atomic_sub_fetch_64(value, -limit);	264✔
54	}
55	tscDebug("sessPerUserUpdateFn updated value:%" PRId64 ", limit:%" PRId64, ref, limit);	10,553✔
56	return code;	10,553✔
57	}
58
59	static int32_t sessConnTimeCheckFn(int64_t value, int64_t limit) {	2,534✔
60	int32_t code = 0;	2,534✔
61
62	int64_t cValue = atomic_load_64(value);	2,534✔
63	int64_t cLimit = atomic_load_64(limit);	2,534✔
64	if (cLimit <= 0) {	2,534✔
65	return code;	2,524✔
66	}
67	int64_t currentTime = taosGetTimestampMs();	10✔
68	if ((cValue + cLimit * 1000) < currentTime) {	10✔
69	code = TSDB_CODE_TSC_SESS_CONN_TIMEOUT;	1✔
70	}
71
72	return code;	10✔
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) {	2,533✔
83	int32_t code = 0;	2,533✔
84
85	int64_t currentTime = taosGetTimestampMs();	2,533✔
86	int64_t cValue = atomic_load_64(value);	2,533✔
87	int64_t cLimit = atomic_load_64(limit);	2,533✔
88	if (cLimit <= 0) {	2,533✔
89	return 0;	2,526✔
90	}
91
92	if ((cValue + cLimit * 1000) < currentTime) {	7✔
93	code = TSDB_CODE_TSC_SESS_CONN_IDLE_TIMEOUT;	1✔
94	}
95	return code;	7✔
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) {	2,532✔
107	int32_t code = 0;	2,532✔
108	int64_t cValue = atomic_load_64(value);	2,532✔
109	int64_t cLimit = atomic_load_64(limit);	2,532✔
110	if (cLimit <= 0) {	2,532✔
111	return code;	2,532✔
112	}
113	if (cValue > cLimit) {	×
114	code = TSDB_CODE_TSC_SESS_MAX_CONCURRENCY_LIMIT;	×
115	}
116	return code;	×
117	}
118
119	static int32_t sessMaxConnCurrencyUpdateFn(int64_t *value, int64_t delta) {	5,968✔
120	int32_t code = 0;	5,968✔
121	int64_t ref = 0;	5,968✔
122	if (delta > 0) {	5,968✔
123	ref = atomic_fetch_add_64(value, delta);	2,532✔
124	} else {
125	ref = atomic_fetch_sub_64(value, -delta);	3,436✔
126	}
127
128	tscDebug("sessMaxConnCurrencyUpdateFn updated value:%" PRId64 ", delta:%" PRId64, ref, delta);	5,968✔
129	return code;	5,968✔
130	}
131
132	static int32_t sessVnodeCallCheckFn(int64_t value, int64_t limit) {	1,884✔
133	int32_t code = 0;	1,884✔
134	int64_t cValue = atomic_load_64(value);	1,884✔
135	int64_t cLimit = atomic_load_64(limit);	1,884✔
136	if (cLimit <= 0) {	1,884✔
137	return code;	1,882✔
138	}
139
140	if (cValue > cLimit) {	2✔
141	code = TSDB_CODE_TSC_SESS_MAX_CALL_VNODE_LIMIT;	1✔
142	}
143	return code;	2✔
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,865✔
158	int32_t code = 0;	5,865✔
159	atomic_store_64(pLimit, src);	5,865✔
160	tscDebug("sessSetValueLimitFn set limit value:%" PRId64, src);	5,865✔
161	return code;	5,865✔
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) {	53✔
173	HANDLE_SESSION_CONTROL();	53✔
174	int32_t code = 0;	53✔
175	SSessMetric pMetric = (SSessMetric )taosMemoryCalloc(1, sizeof(SSessMetric));	53✔
176	if (pMetric == NULL) {	53✔
177	code = terrno;	×
178	return code;	×
179	}
180
181	memset(pMetric->value, 0, sizeof(pMetric->value));	53✔
182
183	pMetric->limit[SESSION_PER_USER] = sessionPerUser;	53✔
184	pMetric->limit[SESSION_CONN_TIME] = sessionConnTime;	53✔
185	pMetric->limit[SESSION_CONN_IDLE_TIME] = sessionConnIdleTime;	53✔
186	pMetric->limit[SESSION_MAX_CONCURRENCY] = sessionMaxConcurrency;	53✔
187	pMetric->limit[SESSION_MAX_CALL_VNODE_NUM] = sessionMaxCallVnodeNum;	53✔
188
189	tstrncpy(pMetric->user, user, sizeof(pMetric->user));	53✔
190	*ppMetric = pMetric;	53✔
191	return code;	53✔
192	}
193
194	int32_t sessMetricUpdateLimit(SSessMetric *pMetric, ESessionType type, int32_t value) {	5,865✔
195	int32_t code = 0;	5,865✔
196
197	code = sessFnSet[type].limitFn(&pMetric->limit[type], value);	5,865✔
198	return code;	5,865✔
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) {	12,831✔
214	return sessFnSet[type].checkFn(&pMetric->value[type], &pMetric->limit[type]);	12,831✔
215	}
216
217	int32_t sessMetricUpdate(SSessMetric pMetric, SSessParam p) {	16,531✔
218	int32_t code = 0;	16,531✔
219	int32_t lino = 0;	16,531✔
220
221	if (p->noCheck == 0) {	16,531✔
222	code = sessMetricCheckByTypeImpl(pMetric, p->type);	12,831✔
223	TAOS_CHECK_GOTO(code, &lino, _error);	12,831✔
224	}
225
226	code = sessFnSet[p->type].updateFn(&pMetric->value[p->type], p->value);	16,521✔
227	_error:	16,531✔
228	return code;	16,531✔
229	}
230
231	int32_t sessMetricCheckValue(SSessMetric *pMetric, ESessionType type, int64_t value) {	6,951✔
232	int32_t code = 0;	6,951✔
233	code = sessFnSet[type].checkFn(&value, &pMetric->limit[type]);	6,951✔
234	return code;	6,951✔
235	}
236	void sessMetricDestroy(SSessMetric *pMetric) { taosMemoryFree(pMetric); }	51✔
237	void sessMetricRef(SSessMetric *pMetric) { TAOS_UNUSED(atomic_add_fetch_32(&pMetric->refCnt, 1)); }	279✔
238	int32_t sessMetricUnref(SSessMetric *pMetric) {	264✔
239	int32_t ref = atomic_sub_fetch_32(&pMetric->refCnt, 1);	264✔
240	if (ref == 0) {	264✔
241	tscDebug("sessMetricUnref, destroy sess metric for user:%s", pMetric->user);	136✔
242	}
243	return ref;	264✔
244	}
245
246	int32_t sessMgtInit() {	50✔
247	int32_t code = 0;	50✔
248	int32_t lino = 0;	50✔
249
250	sessMgt.pSessMetricMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);	50✔
251	if (sessMgt.pSessMetricMap == NULL) {	50✔
252	code = terrno;	×
253	TAOS_CHECK_GOTO(code, &lino, _error);	×
254	}
255	sessMgt.inited = 1;	50✔
256
257	_error:	50✔
258	if (code != 0) {	50✔
259	tscError("failed to init session mgt, line:%d, code:%d", lino, code);	×
260	}
261	return code;	50✔
262	}
263
264	int32_t sessMgtGetOrCreateUserMetric(char user, SSessMetric *pMetric) {	279✔
265	HANDLE_SESSION_CONTROL();	279✔
266	int32_t lino = 0;	279✔
267	int32_t code = 0;	279✔
268
269	SSessMetric *p = NULL;	279✔
270
271	(void)taosThreadRwlockWrlock(&sessMgt.lock);	279✔
272	SSessMetric **ppMetric = taosHashGet(sessMgt.pSessMetricMap, user, strlen(user));	279✔
273	if (ppMetric == NULL \|\| *ppMetric == NULL) {	279✔
274	code = sessMetricCreate(user, &p);	43✔
275	TAOS_CHECK_GOTO(code, &lino, _error);	43✔
276
277	code = taosHashPut(sessMgt.pSessMetricMap, user, strlen(user), &p, sizeof(SSessMetric *));	43✔
278	TAOS_CHECK_GOTO(code, &lino, _error);	43✔
279	} else {
280	p = *ppMetric;	236✔
281	}
282
283	*pMetric = p;	279✔
284
285	_error:	279✔
286	(void)taosThreadRwlockUnlock(&sessMgt.lock);	279✔
287	return code;	279✔
288	}
289
290	int32_t sessMgtUpdataLimit(char *user, ESessionType type, int32_t value) {	5,875✔
291	HANDLE_SESSION_CONTROL();	5,875✔
292	int32_t code = 0;	5,875✔
293	int32_t lino = 0;	5,875✔
294	SSessionMgt *pMgt = &sessMgt;	5,875✔
295
296	if (type >= SESSION_MAX_TYPE \|\| type < SESSION_PER_USER) {	5,875✔
297	return TSDB_CODE_INVALID_PARA;	10✔
298	}
299
300	SSessMetric *pMetric = NULL;	5,865✔
301	(void)taosThreadRwlockWrlock(&pMgt->lock);	5,865✔
302
303	SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));	5,865✔
304	if (ppMetric == NULL \|\| *ppMetric == NULL) {	5,865✔
305	code = sessMetricCreate(user, &pMetric);	10✔
306	TAOS_CHECK_GOTO(code, &lino, _error);	10✔
307
308	code = taosHashPut(pMgt->pSessMetricMap, user, strlen(user), &pMetric, sizeof(SSessMetric *));	10✔
309	TAOS_CHECK_GOTO(code, &lino, _error);	10✔
310
311	} else {
312	pMetric = *ppMetric;	5,855✔
313	}
314
315	code = sessMetricUpdateLimit(pMetric, type, value);	5,865✔
316	TAOS_CHECK_GOTO(code, &lino, _error);	5,865✔
317
318	_error:	5,865✔
319	if (code != 0) {	5,865✔
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,865✔
324
325	return code;	5,865✔
326	}
327
328	int32_t sessMgtUpdateUserMetric(char user, SSessParam pPara) {	10,020✔
329	HANDLE_SESSION_CONTROL();	10,020✔
330
331	int32_t code = 0;	10,020✔
332	int32_t lino = 0;	10,020✔
333
334	SSessionMgt *pMgt = &sessMgt;	10,020✔
335
336	SSessMetric *pMetric = NULL;	10,020✔
337	(void)taosThreadRwlockWrlock(&pMgt->lock);	10,020✔
338
339	SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));	10,020✔
340	if (ppMetric == NULL \|\| *ppMetric == NULL) {	10,020✔
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;	10,020✔
348	}
349
350	code = sessMetricUpdate(pMetric, pPara);	10,020✔
351	TAOS_CHECK_GOTO(code, &lino, _error);	10,020✔
352
353	_error:	10,010✔
354	if (code != 0) {	10,020✔
355	uError("failed to update user session metric, line:%d, code:%d", lino, code);	10✔
356	}
357
358	TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));	10,020✔
359	return code;	10,020✔
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() {	50✔
386	SSessionMgt *pMgt = &sessMgt;	50✔
387	int32_t code = 0;	50✔
388
389	if (pMgt->pSessMetricMap == NULL) {	50✔
390	return;	2✔
391	}
392
393	void *p = taosHashIterate(pMgt->pSessMetricMap, NULL);	48✔
394	while (p) {	99✔
395	SSessMetric pMetric = (SSessMetric **)p;	51✔
396	sessMetricDestroy(pMetric);	51✔
397	p = taosHashIterate(pMgt->pSessMetricMap, p);	51✔
398	}
399
400	code = taosThreadRwlockDestroy(&pMgt->lock);	48✔
401	if (code != 0) {	48✔
402	uError("failed to destroy session mgt, code:%d", code);	×
403	}
404	taosHashCleanup(pMgt->pSessMetricMap);	48✔
405
406	pMgt->pSessMetricMap = NULL;	48✔
407	}
408	int32_t tscCheckConnStatus(STscObj *pTsc) {	2,534✔
409	HANDLE_SESSION_CONTROL();	2,534✔
410
411	int32_t code = 0;	2,534✔
412	int32_t lino = 0;	2,534✔
413
414	SConnAccessInfo *p = &pTsc->sessInfo;	2,534✔
415	SSessMetric pMetric = (SSessMetric )pTsc->pSessMetric;	2,534✔
416
417	code = sessMetricCheckValue(pMetric, SESSION_CONN_TIME, p->startTime);	2,534✔
418	TAOS_CHECK_GOTO(code, &lino, _error);	2,534✔
419
420	code = sessMetricCheckValue(pMetric, SESSION_CONN_IDLE_TIME, p->lastAccessTime);	2,533✔
421	TAOS_CHECK_GOTO(code, &lino, _error);	2,533✔
422
423	_error:	2,532✔
424	if (code != 0) {	2,534✔
425	uError("failed to check connection status line:%d, code:%d", lino, code);	2✔
426	}
427	return code;	2,534✔
428	}
429
430	int32_t connCheckAndUpateMetric(int64_t connId) {	2,534✔
431	HANDLE_SESSION_CONTROL();	2,534✔
432
433	int32_t code = 0;	2,534✔
434	int32_t lino = 0;	2,534✔
435
436	STscObj *pTscObj = acquireTscObj(connId);	2,534✔
437	if (pTscObj == NULL) {	2,534✔
438	code = TSDB_CODE_INVALID_PARA;	×
439	return code;	×
440	}
441	if (pTscObj->pSessMetric == NULL) {	2,534✔
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);	2,534✔
449	TAOS_CHECK_GOTO(code, &lino, _error);	2,534✔
450
451	updateConnAccessInfo(&pTscObj->sessInfo);	2,532✔
452
453	code =
454	sessMetricUpdate((SSessMetric *)pTscObj->pSessMetric, &(SSessParam){.type = SESSION_MAX_CONCURRENCY, .value = 1});	2,532✔
455	TAOS_CHECK_GOTO(code, &lino, _error);	2,532✔
456
457	_error:	2,532✔
458	if (code != 0) {	2,534✔
459	tscError("conn:0x%" PRIx64 ", check and update metric failed at line:%d, code:%s", connId, lino, tstrerror(code));	2✔
460	}
461
462	releaseTscObj(connId);	2,534✔
463	return code;	2,534✔
464	}
465
466	int32_t tscUpdateSessMetric(STscObj pTscObj, SSessParam pParam) {	4,001✔
467	HANDLE_SESSION_CONTROL();	4,001✔
468
469	int32_t code = 0;	4,001✔
470	int32_t lino = 0;	4,001✔
471
472	if (pTscObj == NULL) {	4,001✔
473	code = TSDB_CODE_INVALID_PARA;	×
474	return code;	×
475	}
476
477	SSessMetric pMetric = (SSessMetric )pTscObj->pSessMetric;	4,001✔
478	if (pTscObj->pSessMetric == NULL) {	4,001✔
479	return code;	22✔
480	}
481
482	code = sessMetricUpdate(pMetric, pParam);	3,979✔
483
484	TAOS_CHECK_GOTO(code, &lino, _error);	3,979✔
485
486	_error:	3,979✔
487
488	return code;	3,979✔
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) {	279✔
509	HANDLE_SESSION_CONTROL();	279✔
510
511	int32_t code = 0;	279✔
512	int32_t lino = 0;	279✔
513
514	SSessMetric *pMetric = NULL;	279✔
515	if (pTscObj->pSessMetric != NULL) {	279✔
516	return TSDB_CODE_SUCCESS;	×
517	}
518	code = sessMgtGetOrCreateUserMetric((char *)pTscObj->user, &pMetric);	279✔
519	TAOS_CHECK_GOTO(code, &lino, _error);	279✔
520
521	pTscObj->pSessMetric = pMetric;	279✔
522	_error:	279✔
523	return code;	279✔
524	}
525	int32_t tscUnrefSessMetric(STscObj *pTscObj) {	271✔
526	HANDLE_SESSION_CONTROL();	271✔
527	int32_t code = 0;	271✔
528
529	int32_t ref = 0;	271✔
530	SSessMetric pMetric = (SSessMetric )pTscObj->pSessMetric;	271✔
531	if (pMetric != NULL) {	271✔
532	ref = sessMetricUnref(pMetric);	264✔
533	tscDebug("tscUnrefSessMetric conn:0x%" PRIx64 ", sess metric ref:%d", pTscObj->id, ref);	264✔
534	pTscObj->pSessMetric = NULL;	264✔
535	}
536	return code;	271✔
537	}

taosdata / TDengine / #4986

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