#4906

Committed 30 Dec 2025 10:52AM UTC coverage: 65.514% (+0.09%) from 65.423%

Build # #4906

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

enh: drop multi-stream (#33962)

Run Details

60 of 106 new or added lines in 4 files covered. (56.6%)

4080 existing lines in 123 files now uncovered.

193840 of 295877 relevant lines covered (65.51%)

120444601.14 hits per line

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

68.32

/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,521,108✔
33	int32_t code = 0;	2,521,108✔
34	int64_t cValue = atomic_load_64(value);	2,521,108✔
35	int64_t cLimit = atomic_load_64(limit);	2,521,108✔
36	if (cLimit <= 0) {	2,521,108✔
37	return 0;	2,518,220✔
38	}
39
40	if (cValue > cLimit) {	2,888✔
UNCOV 41	code = TSDB_CODE_TSC_SESS_PER_USER_LIMIT;	×
42	}
43
44	return code;	2,888✔
45	}
46
47	static int32_t sessPerUserUpdateFn(int64_t *value, int64_t limit) {	5,037,724✔
48	int32_t code = 0;	5,037,724✔
49	int64_t ref = 0;	5,037,724✔
50	if (limit > 0) {	5,037,724✔
51	ref = atomic_add_fetch_64(value, limit);	2,521,121✔
52	} else {
53	ref = atomic_sub_fetch_64(value, -limit);	2,516,603✔
54	}
55	tscDebug("sessPerUserUpdateFn updated value:%" PRId64 ", limit:%" PRId64, ref, limit);	5,037,774✔
56	return code;	5,037,774✔
57	}
58
59	static int32_t sessConnTimeCheckFn(int64_t value, int64_t limit) {	785,901,481✔
60	int32_t code = 0;	785,901,481✔
61
62	int64_t cValue = atomic_load_64(value);	785,901,481✔
63	int64_t cLimit = atomic_load_64(limit);	785,901,408✔
64	if (cLimit <= 0) {	785,903,519✔
65	return code;	5,181,691✔
66	}
67	int64_t currentTime = taosGetTimestampMs();	780,720,111✔
68	if ((cValue + cLimit * 1000) < currentTime) {	780,720,111✔
69	code = TSDB_CODE_TSC_SESS_CONN_TIMEOUT;	×
70	}
71
72	return code;	780,720,111✔
73	}
74
75	static int32_t sessConnTimeUpdateFn(int64_t *value, int64_t limit) {	×
UNCOV 76	int32_t code = 0;	×
77	int64_t now = taosGetTimestampMs();	×
UNCOV 78	atomic_store_64(value, now);	×
UNCOV 79	return code;	×
80	}
81
82	static int32_t sessConnIdleTimeCheckFn(int64_t value, int64_t limit) {	785,900,475✔
83	int32_t code = 0;	785,900,475✔
84
85	int64_t currentTime = taosGetTimestampMs();	785,907,669✔
86	int64_t cValue = atomic_load_64(value);	785,907,669✔
87	int64_t cLimit = atomic_load_64(limit);	785,900,962✔
88	if (cLimit <= 0) {	785,905,849✔
89	return 0;	5,181,695✔
90	}
91
92	if ((cValue + cLimit * 1000) < currentTime) {	780,724,154✔
UNCOV 93	code = TSDB_CODE_TSC_SESS_CONN_IDLE_TIMEOUT;	×
94	}
95	return code;	780,724,154✔
96	}
97
UNCOV 98	static int32_t sessConnIdleTimeUpdateFn(int64_t *value, int64_t limit) {	×
99	int32_t code = 0;	×
UNCOV 100	int64_t now = taosGetTimestampMs();	×
UNCOV 101	atomic_store_64(value, now);	×
102
103	return code;	×
104	}
105
106	static int32_t sessMaxConnCurrencyCheckFn(int64_t value, int64_t limit) {	785,899,554✔
107	int32_t code = 0;	785,899,554✔
108	int64_t cValue = atomic_load_64(value);	785,899,554✔
109	int64_t cLimit = atomic_load_64(limit);	785,901,438✔
110	if (cLimit <= 0) {	785,904,815✔
111	return code;	785,298,524✔
112	}
113	if (cValue > cLimit) {	606,291✔
114	code = TSDB_CODE_TSC_SESS_MAX_CONCURRENCY_LIMIT;	×
115	}
116	return code;	606,291✔
117	}
118
119	static int32_t sessMaxConnCurrencyUpdateFn(int64_t *value, int64_t delta) {	1,578,434,941✔
120	int32_t code = 0;	1,578,434,941✔
121	int64_t ref = 0;	1,578,434,941✔
122	if (delta > 0) {	1,578,434,941✔
123	ref = atomic_fetch_add_64(value, delta);	785,903,987✔
124	} else {
125	ref = atomic_fetch_sub_64(value, -delta);	792,530,954✔
126	}
127
128	tscDebug("sessMaxConnCurrencyUpdateFn updated value:%" PRId64 ", delta:%" PRId64, ref, delta);	1,578,453,353✔
129	return code;	1,578,453,360✔
130	}
131
132	static int32_t sessVnodeCallCheckFn(int64_t value, int64_t limit) {	707,838,256✔
133	int32_t code = 0;	707,838,256✔
134	int64_t cValue = atomic_load_64(value);	707,838,256✔
135	int64_t cLimit = atomic_load_64(limit);	707,840,407✔
136	if (cLimit <= 0) {	707,841,657✔
137	return code;	707,841,657✔
138	}
139
UNCOV 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;	×
UNCOV 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	}
UNCOV 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) {	6,351,090✔
158	int32_t code = 0;	6,351,090✔
159	atomic_store_64(pLimit, src);	6,351,090✔
160	tscDebug("sessSetValueLimitFn set limit value:%" PRId64, src);	6,351,090✔
161	return code;	6,351,090✔
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,620,942✔
173	HANDLE_SESSION_CONTROL();	1,620,942✔
174	int32_t code = 0;	1,620,942✔
175	SSessMetric pMetric = (SSessMetric )taosMemoryCalloc(1, sizeof(SSessMetric));	1,620,942✔
176	if (pMetric == NULL) {	1,620,942✔
UNCOV 177	code = terrno;	×
178	return code;	×
179	}
180
181	memset(pMetric->value, 0, sizeof(pMetric->value));	1,620,942✔
182
183	pMetric->limit[SESSION_PER_USER] = sessionPerUser;	1,620,942✔
184	pMetric->limit[SESSION_CONN_TIME] = sessionConnTime;	1,620,942✔
185	pMetric->limit[SESSION_CONN_IDLE_TIME] = sessionConnIdleTime;	1,620,942✔
186	pMetric->limit[SESSION_MAX_CONCURRENCY] = sessionMaxConcurrency;	1,620,942✔
187	pMetric->limit[SESSION_MAX_CALL_VNODE_NUM] = sessionMaxCallVnodeNum;	1,620,942✔
188
189	tstrncpy(pMetric->user, user, sizeof(pMetric->user));	1,620,942✔
190	*ppMetric = pMetric;	1,620,942✔
191	return code;	1,620,942✔
192	}
193
194	int32_t sessMetricUpdateLimit(SSessMetric *pMetric, ESessionType type, int32_t value) {	6,351,090✔
195	int32_t code = 0;	6,351,090✔
196
197	code = sessFnSet[type].limitFn(&pMetric->limit[type], value);	6,351,090✔
198	return code;	6,351,090✔
199	}
200
UNCOV 201	int32_t sessMetricCheckImpl(SSessMetric *pMetric) {	×
202	int32_t code = 0;	×
203
UNCOV 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) {	×
UNCOV 207	break;	×
208	}
209	}
210
UNCOV 211	return code;	×
212	}
213	int32_t sessMetricCheckByTypeImpl(SSessMetric *pMetric, ESessionType type) {	788,420,176✔
214	return sessFnSet[type].checkFn(&pMetric->value[type], &pMetric->limit[type]);	788,420,176✔
215	}
216
217	int32_t sessMetricUpdate(SSessMetric pMetric, SSessParam p) {	1,583,470,849✔
218	int32_t code = 0;	1,583,470,849✔
219	int32_t lino = 0;	1,583,470,849✔
220
221	if (p->noCheck == 0) {	1,583,476,715✔
222	code = sessMetricCheckByTypeImpl(pMetric, p->type);	788,426,196✔
223	TAOS_CHECK_GOTO(code, &lino, _error);	788,426,777✔
224	}
225
226	code = sessFnSet[p->type].updateFn(&pMetric->value[p->type], p->value);	1,583,482,551✔
227	_error:	1,583,491,075✔
228	return code;	1,583,491,075✔
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,620,942✔
237	void sessMetricRef(SSessMetric *pMetric) { TAOS_UNUSED(atomic_add_fetch_32(&pMetric->refCnt, 1)); }	2,521,121✔
238	int32_t sessMetricUnref(SSessMetric *pMetric) {	2,516,616✔
239	int32_t ref = atomic_sub_fetch_32(&pMetric->refCnt, 1);	2,516,616✔
240	if (ref == 0) {	2,516,653✔
241	tscDebug("sessMetricUnref, destroy sess metric for user:%s", pMetric->user);	1,897,271✔
242	}
243	return ref;	2,516,653✔
244	}
245
246	int32_t sessMgtInit() {	1,609,126✔
247	int32_t code = 0;	1,609,126✔
248	int32_t lino = 0;	1,609,126✔
249
250	sessMgt.pSessMetricMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);	1,609,126✔
251	if (sessMgt.pSessMetricMap == NULL) {	1,609,126✔
UNCOV 252	code = terrno;	×
253	TAOS_CHECK_GOTO(code, &lino, _error);	×
254	}
255	sessMgt.inited = 1;	1,609,126✔
256
257	_error:	1,609,126✔
258	if (code != 0) {	1,609,126✔
259	tscError("failed to init session mgt, line:%d, code:%d", lino, code);	×
260	}
261	return code;	1,609,126✔
262	}
263
264	int32_t sessMgtGetOrCreateUserMetric(char user, SSessMetric *pMetric) {	2,520,943✔
265	HANDLE_SESSION_CONTROL();	2,520,943✔
266	int32_t lino = 0;	2,520,943✔
267	int32_t code = 0;	2,520,943✔
268
269	SSessMetric *p = NULL;	2,520,943✔
270
271	(void)taosThreadRwlockWrlock(&sessMgt.lock);	2,520,386✔
272	SSessMetric **ppMetric = taosHashGet(sessMgt.pSessMetricMap, user, strlen(user));	2,521,121✔
273	if (ppMetric == NULL \|\| *ppMetric == NULL) {	2,521,121✔
274	code = sessMetricCreate(user, &p);	1,620,942✔
275	TAOS_CHECK_GOTO(code, &lino, _error);	1,620,942✔
276
277	code = taosHashPut(sessMgt.pSessMetricMap, user, strlen(user), &p, sizeof(SSessMetric *));	1,620,942✔
278	TAOS_CHECK_GOTO(code, &lino, _error);	1,620,942✔
279	} else {
280	p = *ppMetric;	900,179✔
281	}
282
283	*pMetric = p;	2,521,121✔
284
285	_error:	2,521,121✔
286	(void)taosThreadRwlockUnlock(&sessMgt.lock);	2,521,121✔
287	return code;	2,521,121✔
288	}
289
290	int32_t sessMgtUpdataLimit(char *user, ESessionType type, int32_t value) {	6,351,090✔
291	HANDLE_SESSION_CONTROL();	6,351,090✔
292	int32_t code = 0;	6,351,090✔
293	int32_t lino = 0;	6,351,090✔
294	SSessionMgt *pMgt = &sessMgt;	6,351,090✔
295
296	if (type >= SESSION_MAX_TYPE \|\| type < SESSION_PER_USER) {	6,351,090✔
UNCOV 297	return TSDB_CODE_INVALID_PARA;	×
298	}
299
300	SSessMetric *pMetric = NULL;	6,351,090✔
301	(void)taosThreadRwlockWrlock(&pMgt->lock);	6,351,090✔
302
303	SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));	6,351,090✔
304	if (ppMetric == NULL \|\| *ppMetric == NULL) {	6,351,090✔
305	code = sessMetricCreate(user, &pMetric);	×
UNCOV 306	TAOS_CHECK_GOTO(code, &lino, _error);	×
307
UNCOV 308	code = taosHashPut(pMgt->pSessMetricMap, user, strlen(user), &pMetric, sizeof(SSessMetric *));	×
309	TAOS_CHECK_GOTO(code, &lino, _error);	×
310
311	} else {
312	pMetric = *ppMetric;	6,351,090✔
313	}
314
315	code = sessMetricUpdateLimit(pMetric, type, value);	6,351,090✔
316	TAOS_CHECK_GOTO(code, &lino, _error);	6,351,090✔
317
318	_error:	6,351,090✔
319	if (code != 0) {	6,351,090✔
320	uError("failed to update session mgt type:%d, line:%d, code:%d", type, lino, code);	×
321	}
322
323	TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));	6,351,090✔
324
325	return code;	6,351,090✔
326	}
327
328	int32_t sessMgtUpdateUserMetric(char user, SSessParam pPara) {	×
UNCOV 329	HANDLE_SESSION_CONTROL();	×
330
331	int32_t code = 0;	×
332	int32_t lino = 0;	×
333
UNCOV 334	SSessionMgt *pMgt = &sessMgt;	×
335
336	SSessMetric *pMetric = NULL;	×
337	(void)taosThreadRwlockWrlock(&pMgt->lock);	×
338
339	SSessMetric **ppMetric = taosHashGet(pMgt->pSessMetricMap, user, strlen(user));	×
UNCOV 340	if (ppMetric == NULL \|\| *ppMetric == NULL) {	×
341	code = sessMetricCreate(user, &pMetric);	×
342	TAOS_CHECK_GOTO(code, &lino, _error);	×
343
UNCOV 344	code = taosHashPut(pMgt->pSessMetricMap, user, strlen(user), &pMetric, sizeof(SSessMetric *));	×
345	TAOS_CHECK_GOTO(code, &lino, _error);	×
346	} else {
UNCOV 347	pMetric = *ppMetric;	×
348	}
349
350	code = sessMetricUpdate(pMetric, pPara);	×
351	TAOS_CHECK_GOTO(code, &lino, _error);	×
352
UNCOV 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));	×
UNCOV 359	return code;	×
360	}
361
UNCOV 362	int32_t sessMgtRemoveUser(char *user) {	×
363	HANDLE_SESSION_CONTROL();	×
364
365	int32_t code = 0;	×
366	int32_t lino = 0;	×
UNCOV 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) {	×
UNCOV 374	if (*ppMetric != NULL) {	×
375	SSessMetric p = ppMetric;	×
376	TAOS_UNUSED(taosHashRemove(pMgt->pSessMetricMap, user, strlen(user)));	×
377	sessMetricDestroy(p);	×
378	}
379	}
UNCOV 380	_error:	×
381	TAOS_UNUSED(taosThreadRwlockUnlock(&pMgt->lock));	×
UNCOV 382	return code;	×
383	}
384
385	void sessMgtDestroy() {	1,609,142✔
386	SSessionMgt *pMgt = &sessMgt;	1,609,142✔
387	int32_t code = 0;	1,609,142✔
388
389	if (pMgt->pSessMetricMap == NULL) {	1,609,142✔
390	return;	16✔
391	}
392
393	void *p = taosHashIterate(pMgt->pSessMetricMap, NULL);	1,609,126✔
394	while (p) {	3,230,068✔
395	SSessMetric pMetric = (SSessMetric **)p;	1,620,942✔
396	sessMetricDestroy(pMetric);	1,620,942✔
397	p = taosHashIterate(pMgt->pSessMetricMap, p);	1,620,942✔
398	}
399
400	code = taosThreadRwlockDestroy(&pMgt->lock);	1,609,126✔
401	if (code != 0) {	1,609,126✔
UNCOV 402	uError("failed to destroy session mgt, code:%d", code);	×
403	}
404	taosHashCleanup(pMgt->pSessMetricMap);	1,609,126✔
405
406	pMgt->pSessMetricMap = NULL;	1,609,126✔
407	}
408	int32_t tscCheckConnStatus(STscObj *pTsc) {	785,901,920✔
409	HANDLE_SESSION_CONTROL();	785,901,920✔
410
411	int32_t code = 0;	785,901,920✔
412	int32_t lino = 0;	785,901,920✔
413
414	SConnAccessInfo *p = &pTsc->sessInfo;	785,903,645✔
415	SSessMetric pMetric = (SSessMetric )pTsc->pSessMetric;	785,904,730✔
416
417	code = sessMetricCheckValue(pMetric, SESSION_CONN_TIME, p->startTime);	785,905,933✔
418	TAOS_CHECK_GOTO(code, &lino, _error);	785,902,645✔
419
420	code = sessMetricCheckValue(pMetric, SESSION_CONN_IDLE_TIME, p->lastAccessTime);	785,902,645✔
421	TAOS_CHECK_GOTO(code, &lino, _error);	785,905,266✔
422
423	_error:	785,905,266✔
424	if (code != 0) {	785,906,073✔
425	uError("failed to check connection status line:%d, code:%d", lino, code);	×
426	}
427	return code;	785,900,479✔
428	}
429
430	int32_t connCheckAndUpateMetric(int64_t connId) {	785,901,381✔
431	HANDLE_SESSION_CONTROL();	785,901,381✔
432
433	int32_t code = 0;	785,901,381✔
434	int32_t lino = 0;	785,901,381✔
435
436	STscObj *pTscObj = acquireTscObj(connId);	785,903,022✔
437	if (pTscObj == NULL) {	785,907,813✔
UNCOV 438	code = TSDB_CODE_INVALID_PARA;	×
UNCOV 439	return code;	×
440	}
441	if (pTscObj->pSessMetric == NULL) {	785,907,813✔
UNCOV 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);	785,907,813✔
449	TAOS_CHECK_GOTO(code, &lino, _error);	785,900,792✔
450
451	updateConnAccessInfo(&pTscObj->sessInfo);	785,900,792✔
452
453	code =
454	sessMetricUpdate((SSessMetric *)pTscObj->pSessMetric, &(SSessParam){.type = SESSION_MAX_CONCURRENCY, .value = 1});	785,902,316✔
455	TAOS_CHECK_GOTO(code, &lino, _error);	785,906,832✔
456
457	_error:	785,906,832✔
458	if (code != 0) {	785,907,261✔
459	tscError("conn:0x%" PRIx64 ", check and update metric failed at line:%d, code:%s", connId, lino, tstrerror(code));	×
460	}
461
462	releaseTscObj(connId);	785,907,261✔
463	return code;	785,901,048✔
464	}
465
466	int32_t tscUpdateSessMetric(STscObj pTscObj, SSessParam pParam) {	797,643,556✔
467	HANDLE_SESSION_CONTROL();	797,643,556✔
468
469	int32_t code = 0;	797,643,556✔
470	int32_t lino = 0;	797,643,556✔
471
472	if (pTscObj == NULL) {	797,645,079✔
473	code = TSDB_CODE_INVALID_PARA;	×
UNCOV 474	return code;	×
475	}
476
477	SSessMetric pMetric = (SSessMetric )pTscObj->pSessMetric;	797,645,079✔
478	if (pTscObj->pSessMetric == NULL) {	797,646,531✔
479	return code;	65,392✔
480	}
481
482	code = sessMetricUpdate(pMetric, pParam);	797,580,643✔
483
484	TAOS_CHECK_GOTO(code, &lino, _error);	797,583,447✔
485
486	_error:	797,583,447✔
487
488	return code;	797,583,451✔
489	}
490
491	int32_t tscCheckConnSessionMetric(STscObj *pTscObj) {	×
UNCOV 492	HANDLE_SESSION_CONTROL();	×
493
UNCOV 494	int32_t code = 0;	×
495	if (pTscObj == NULL) {	×
496	code = TSDB_CODE_INVALID_PARA;	×
UNCOV 497	return code;	×
498	}
499	code = tscCheckConnStatus(pTscObj);	×
500
UNCOV 501	_error:	×
UNCOV 502	if (code != 0) {	×
503	uError("failed to check connection session metric, code:%d", code);	×
504	}
UNCOV 505	return code;	×
506	}
507
508	int32_t tscRefSessMetric(STscObj *pTscObj) {	2,520,915✔
509	HANDLE_SESSION_CONTROL();	2,520,915✔
510
511	int32_t code = 0;	2,520,915✔
512	int32_t lino = 0;	2,520,915✔
513
514	SSessMetric *pMetric = NULL;	2,520,915✔
515	if (pTscObj->pSessMetric != NULL) {	2,520,915✔
516	return TSDB_CODE_SUCCESS;	×
517	}
518	code = sessMgtGetOrCreateUserMetric((char *)pTscObj->user, &pMetric);	2,521,010✔
519	TAOS_CHECK_GOTO(code, &lino, _error);	2,521,121✔
520
521	pTscObj->pSessMetric = pMetric;	2,521,121✔
522	_error:	2,521,121✔
523	return code;	2,521,121✔
524	}
525	int32_t tscUnrefSessMetric(STscObj *pTscObj) {	2,517,572✔
526	HANDLE_SESSION_CONTROL();	2,517,572✔
527	int32_t code = 0;	2,517,572✔
528
529	int32_t ref = 0;	2,517,572✔
530	SSessMetric pMetric = (SSessMetric )pTscObj->pSessMetric;	2,517,572✔
531	if (pMetric != NULL) {	2,517,609✔
532	ref = sessMetricUnref(pMetric);	2,516,653✔
533	tscDebug("tscUnrefSessMetric conn:0x%" PRIx64 ", sess metric ref:%d", pTscObj->id, ref);	2,516,653✔
534	pTscObj->pSessMetric = NULL;	2,516,653✔
535	}
536	return code;	2,517,572✔
537	}

taosdata / TDengine / #4906

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