xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

replace sdb func

This commit is contained in:
Shengliang Guan 2020-05-13 17:24:06 +08:00
parent f3b80c84e5
commit a23b0b0cd8
19 changed files with 69 additions and 1233 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/client/src/tscSub.c
View File

@ -14,8 +14,6 @@
*/ */
#include "os.h" #include "os.h"
#include "shash.h"
#include "taos.h" #include "taos.h"
#include "trpc.h" #include "trpc.h"
#include "tsclient.h" #include "tsclient.h"

2
src/dnode/src/dnodeMgmt.c
View File

@ -16,7 +16,6 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "cJSON.h" #include "cJSON.h"
#include "ihash.h"
#include "taoserror.h" #include "taoserror.h"
#include "taosmsg.h" #include "taosmsg.h"
#include "ttime.h" #include "ttime.h"
@ -174,6 +173,7 @@ static int32_t dnodeOpenVnodes() {
int32_t numOfVnodes; int32_t numOfVnodes;
int32_t status; int32_t status;
vnodeInit();
status = dnodeGetVnodeList(vnodeList, &numOfVnodes); status = dnodeGetVnodeList(vnodeList, &numOfVnodes);
if (status != TSDB_CODE_SUCCESS) { if (status != TSDB_CODE_SUCCESS) {

1
src/inc/vnode.h
View File

@ -35,6 +35,7 @@ typedef struct {
void *qhandle; //used by query and retrieve msg void *qhandle; //used by query and retrieve msg
} SRspRet; } SRspRet;
void vnodeInit();
int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg); int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg);
int32_t vnodeDrop(int32_t vgId); int32_t vnodeDrop(int32_t vgId);
int32_t vnodeOpen(int32_t vgId, char *rootDir); int32_t vnodeOpen(int32_t vgId, char *rootDir);

2
src/mnode/src/mgmtSdb.c
View File

@ -24,8 +24,6 @@
#include "twal.h" #include "twal.h"
#include "tsync.h" #include "tsync.h"
#include "tglobal.h" #include "tglobal.h"
#include "hashint.h"
#include "hashstr.h"
#include "dnode.h" #include "dnode.h"
#include "mgmtDef.h" #include "mgmtDef.h"
#include "mgmtInt.h" #include "mgmtInt.h"

1
src/plugins/http/src/httpHandle.c
View File

@ -19,7 +19,6 @@
#include "tglobal.h" #include "tglobal.h"
#include "tsocket.h" #include "tsocket.h"
#include "ttimer.h" #include "ttimer.h"
#include "shash.h"
#include "http.h" #include "http.h"
#include "httpLog.h" #include "httpLog.h"
#include "httpCode.h" #include "httpCode.h"

69
src/plugins/http/src/httpSession.c
View File

@ -15,7 +15,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "shash.h" #include "hash.h"
#include "taos.h" #include "taos.h"
#include "ttime.h" #include "ttime.h"
#include "ttimer.h" #include "ttimer.h"
@ -25,7 +25,6 @@
#include "httpHandle.h" #include "httpHandle.h"
#include "httpResp.h" #include "httpResp.h"
void httpAccessSession(HttpContext *pContext) { void httpAccessSession(HttpContext *pContext) {
HttpServer *server = pContext->pThread->pServer; HttpServer *server = pContext->pThread->pServer;
pthread_mutex_lock(&server->serverMutex); pthread_mutex_lock(&server->serverMutex);
@ -50,8 +49,11 @@ void httpCreateSession(HttpContext *pContext, void *taos) {
session.taos = taos; session.taos = taos;
session.expire = (int)taosGetTimestampSec() + server->sessionExpire; session.expire = (int)taosGetTimestampSec() + server->sessionExpire;
session.access = 1; session.access = 1;
snprintf(session.id, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass); int sessionIdLen = snprintf(session.id, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass);
pContext->session = (HttpSession *)taosAddStrHash(server->pSessionHash, session.id, (char *)(&session));
taosHashPut(server->pSessionHash, session.id, sessionIdLen, (char *)(&session), sizeof(HttpSession));
pContext->session = taosHashGet(server->pSessionHash, session.id, sessionIdLen);
if (pContext->session == NULL) { if (pContext->session == NULL) {
httpError("context:%p, fd:%d, ip:%s, user:%s, error:%s", pContext, pContext->fd, pContext->ipstr, pContext->user, httpError("context:%p, fd:%d, ip:%s, user:%s, error:%s", pContext, pContext->fd, pContext->ipstr, pContext->user,
httpMsg[HTTP_SESSION_FULL]); httpMsg[HTTP_SESSION_FULL]);
@ -71,9 +73,9 @@ void httpFetchSessionImp(HttpContext *pContext) {
pthread_mutex_lock(&server->serverMutex); pthread_mutex_lock(&server->serverMutex);
char sessionId[HTTP_SESSION_ID_LEN]; char sessionId[HTTP_SESSION_ID_LEN];
snprintf(sessionId, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass); int sessonIdLen = snprintf(sessionId, HTTP_SESSION_ID_LEN, "%s.%s", pContext->user, pContext->pass);
pContext->session = (HttpSession *)taosGetStrHashData(server->pSessionHash, sessionId); pContext->session = taosHashGet(server->pSessionHash, sessionId, sessonIdLen);
if (pContext->session != NULL && pContext->session == pContext->session->signature) { if (pContext->session != NULL && pContext->session == pContext->session->signature) {
pContext->session->access++; pContext->session->access++;
httpTrace("context:%p, fd:%d, ip:%s, user:%s, find an exist session:%p:%p, access:%d, expire:%d", httpTrace("context:%p, fd:%d, ip:%s, user:%s, find an exist session:%p:%p, access:%d, expire:%d",
@ -120,8 +122,7 @@ void httpRestoreSession(HttpContext *pContext) {
pthread_mutex_unlock(&server->serverMutex); pthread_mutex_unlock(&server->serverMutex);
} }
void httpResetSession(char *session) { void httpResetSession(HttpSession *pSession) {
HttpSession *pSession = (HttpSession *)session;
httpTrace("close session:%p:%p", pSession, pSession->taos); httpTrace("close session:%p:%p", pSession, pSession->taos);
if (pSession->taos != NULL) { if (pSession->taos != NULL) {
taos_close(pSession->taos); taos_close(pSession->taos);
@ -131,15 +132,20 @@ void httpResetSession(char *session) {
} }
void httpRemoveAllSessions(HttpServer *pServer) { void httpRemoveAllSessions(HttpServer *pServer) {
if (pServer->pSessionHash != NULL) { SHashMutableIterator *pIter = taosHashCreateIter(pServer->pSessionHash);
taosCleanUpStrHashWithFp(pServer->pSessionHash, httpResetSession);
pServer->pSessionHash = NULL; while (taosHashIterNext(pIter)) {
HttpSession *pSession = taosHashIterGet(pIter);
if (pSession == NULL) continue;
httpResetSession(pSession);
} }
taosHashDestroyIter(pIter);
} }
bool httpInitAllSessions(HttpServer *pServer) { bool httpInitAllSessions(HttpServer *pServer) {
if (pServer->pSessionHash == NULL) { if (pServer->pSessionHash == NULL) {
pServer->pSessionHash = taosInitStrHash(100, sizeof(HttpSession), taosHashStringStep1); pServer->pSessionHash = taosHashInit(10, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true);
} }
if (pServer->pSessionHash == NULL) { if (pServer->pSessionHash == NULL) {
httpError("http init session pool failed"); httpError("http init session pool failed");
@ -152,46 +158,41 @@ bool httpInitAllSessions(HttpServer *pServer) {
return true; return true;
} }
int httpSessionExpired(char *session) { bool httpSessionExpired(HttpSession *pSession) {
HttpSession *pSession = (HttpSession *)session; time_t cur = taosGetTimestampSec();
time_t cur = taosGetTimestampSec();
if (pSession->taos != NULL) { if (pSession->taos != NULL) {
if (pSession->expire > cur) { if (pSession->expire > cur) {
return 0; // un-expired, so return false return false; // un-expired, so return false
} }
if (pSession->access > 0) { if (pSession->access > 0) {
httpTrace("session:%p:%p is expired, but still access:%d", pSession, pSession->taos, httpTrace("session:%p:%p is expired, but still access:%d", pSession, pSession->taos,
pSession->access); pSession->access);
return 0; // still used, so return false return false; // still used, so return false
} }
httpTrace("need close session:%p:%p for it expired, cur:%d, expire:%d, invertal:%d", httpTrace("need close session:%p:%p for it expired, cur:%d, expire:%d, invertal:%d",
pSession, pSession->taos, cur, pSession->expire, cur - pSession->expire); pSession, pSession->taos, cur, pSession->expire, cur - pSession->expire);
} }
return 1; return true;
} }
void httpRemoveExpireSessions(HttpServer *pServer) { void httpRemoveExpireSessions(HttpServer *pServer) {
int expiredNum = 0; SHashMutableIterator *pIter = taosHashCreateIter(pServer->pSessionHash);
do {
while (taosHashIterNext(pIter)) {
HttpSession *pSession = taosHashIterGet(pIter);
if (pSession == NULL) continue;
pthread_mutex_lock(&pServer->serverMutex); pthread_mutex_lock(&pServer->serverMutex);
if (httpSessionExpired(pSession)) {
HttpSession *pSession = (HttpSession *)taosVisitStrHashWithFp(pServer->pSessionHash, httpSessionExpired); httpResetSession(pSession);
if (pSession == NULL) { taosHashRemove(pServer->pSessionHash, pSession->id, strlen(pSession->id));
pthread_mutex_unlock(&pServer->serverMutex);
break;
} }
httpResetSession((char *)pSession);
taosDeleteStrHashNode(pServer->pSessionHash, pSession->id, pSession);
pthread_mutex_unlock(&pServer->serverMutex); pthread_mutex_unlock(&pServer->serverMutex);
}
if (++expiredNum > 10) { taosHashDestroyIter(pIter);
break;
}
} while (true);
} }
void httpProcessSessionExpire(void *handle, void *tmrId) { void httpProcessSessionExpire(void *handle, void *tmrId) {

1
src/plugins/http/src/httpSystem.c
View File

@ -15,7 +15,6 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "shash.h"
#include "taos.h" #include "taos.h"
#include "tglobal.h" #include "tglobal.h"
#include "tsocket.h" #include "tsocket.h"

1
src/plugins/http/src/httpUtil.c
View File

@ -16,7 +16,6 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "tmd5.h" #include "tmd5.h"
#include "shash.h"
#include "taos.h" #include "taos.h"
#include "http.h" #include "http.h"
#include "httpLog.h" #include "httpLog.h"

1
src/plugins/http/src/tgHandle.c
View File

@ -16,7 +16,6 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "tglobal.h" #include "tglobal.h"
#include "shash.h"
#include "taosdef.h" #include "taosdef.h"
#include "taosmsg.h" #include "taosmsg.h"
#include "tgHandle.h" #include "tgHandle.h"

2
src/query/src/tvariant.c
View File

@ -17,7 +17,7 @@
#include "hash.h" #include "hash.h"
#include "hashfunc.h" #include "hashfunc.h"
#include "os.h" #include "os.h"
#include "shash.h" #include "hash.h"
#include "taos.h" #include "taos.h"
#include "taosdef.h" #include "taosdef.h"
#include "tstoken.h" #include "tstoken.h"

26
src/util/inc/hashint.h
View File

@ -1,26 +0,0 @@
/*
* 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/>.
*/
#ifndef _sdb_int_hash_header_
#define _sdb_int_hash_header_
void *sdbOpenIntHash(int maxSessions, int dataSize);
void sdbCloseIntHash(void *handle);
void *sdbAddIntHash(void *handle, void *key, void *pData);
void sdbDeleteIntHash(void *handle, void *key);
void *sdbGetIntHashData(void *handle, void *key);
void *sdbFetchIntHashData(void *handle, void *ptr, void **ppMeta);
#endif

26
src/util/inc/hashstr.h
View File

@ -1,26 +0,0 @@
/*
* 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/>.
*/
#ifndef _sdb_str_hash_header_
#define _sdb_str_hash_header_
void *sdbOpenStrHash(int maxSessions, int dataSize);
void sdbCloseStrHash(void *handle);
void *sdbAddStrHash(void *handle, void *key, void *pData);
void sdbDeleteStrHash(void *handle, void *key);
void *sdbGetStrHashData(void *handle, void *key);
void *sdbFetchStrHashData(void *handle, void *ptr, void **ppMeta);
#endif

47
src/util/inc/ihash.h
View File

@ -1,47 +0,0 @@
/*
* 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/>.
*/
#ifndef TDENGINE_IHASH_H
#define TDENGINE_IHASH_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
void *taosInitIntHash(int32_t maxSessions, int32_t dataSize, int32_t (*fp)(void *, uint64_t));
void taosCleanUpIntHash(void *handle);
char *taosGetIntHashData(void *handle, uint64_t key);
void taosDeleteIntHash(void *handle, uint64_t key);
char *taosAddIntHash(void *handle, uint64_t key, char *pData);
int32_t taosHashInt(void *handle, uint64_t key);
void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *));
void taosVisitIntHashWithFp(void *handle, void (*fp)(char *, void *), void *param);
int32_t taosGetIntHashSize(void *handle);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_IHASH_H

51
src/util/inc/shash.h
View File

@ -1,51 +0,0 @@
/*
* 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/>.
*/
#ifndef TDENGINE_TSHASH_H
#define TDENGINE_TSHASH_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
void *taosInitStrHash(uint32_t maxSessions, uint32_t dataSize, uint32_t (*fp)(void *, char *));
void taosCleanUpStrHash(void *handle);
void *taosGetStrHashData(void *handle, char *string);
void taosDeleteStrHash(void *handle, char *string);
void taosDeleteStrHashNode(void *handle, char *string, void *pDeleteNode);
void *taosAddStrHash(void *handle, char *string, char *pData);
void *taosAddStrHashWithSize(void *handle, char *string, char *pData, int dataSize);
uint32_t taosHashString(void *handle, char *string);
uint32_t taosHashStringStep1(void *handle, char *string);
char *taosVisitStrHashWithFp(void *handle, int (*fp)(char *));
void taosCleanUpStrHashWithFp(void *handle, void (*fp)(char *));
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_TSHASH_H

201
src/util/src/hashint.c
View File

@ -1,201 +0,0 @@
/*
* 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 <pthread.h>
#include <semaphore.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "tmempool.h"
#include "taosdef.h"
typedef struct _long_hash_t {
uint32_t key;
int hash;
struct _long_hash_t *prev;
struct _long_hash_t *next;
char data[];
} SLongHash;
typedef struct {
SLongHash **longHashList;
mpool_h longHashMemPool;
int maxSessions;
int dataSize;
} SHashObj;
int sdbHashLong(void *handle, uint32_t ip) {
SHashObj *pObj = (SHashObj *)handle;
int hash = 0;
hash = ip >> 16;
hash += (ip & 0xFFFF);
hash = hash % pObj->maxSessions;
return hash;
}
void *sdbAddIntHash(void *handle, void *pKey, void *data) {
int hash;
SLongHash *pNode;
SHashObj * pObj;
uint32_t key = *((uint32_t *)pKey);
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = sdbHashLong(pObj, key);
pNode = (SLongHash *)taosMemPoolMalloc(pObj->longHashMemPool);
pNode->key = key;
memcpy(pNode->data, data, pObj->dataSize);
pNode->prev = 0;
pNode->next = pObj->longHashList[hash];
pNode->hash = hash;
if (pObj->longHashList[hash] != 0) (pObj->longHashList[hash])->prev = pNode;
pObj->longHashList[hash] = pNode;
return pObj;
}
void sdbDeleteIntHash(void *handle, void *pKey) {
int hash;
SLongHash *pNode;
SHashObj * pObj;
uint32_t key = *((uint32_t *)pKey);
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
hash = sdbHashLong(pObj, key);
pNode = pObj->longHashList[hash];
while (pNode) {
if (pNode->key == key) break;
pNode = pNode->next;
}
if (pNode) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->longHashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
taosMemPoolFree(pObj->longHashMemPool, (char *)pNode);
}
}
void *sdbGetIntHashData(void *handle, void *pKey) {
int hash;
SLongHash *pNode;
SHashObj * pObj;
uint32_t key = *((uint32_t *)pKey);
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = sdbHashLong(pObj, key);
pNode = pObj->longHashList[hash];
while (pNode) {
if (pNode->key == key) {
break;
}
pNode = pNode->next;
}
if (pNode) return pNode->data;
return NULL;
}
void *sdbOpenIntHash(int maxSessions, int dataSize) {
SLongHash **longHashList;
mpool_h longHashMemPool;
SHashObj * pObj;
longHashMemPool = taosMemPoolInit(maxSessions, sizeof(SLongHash) + dataSize);
if (longHashMemPool == 0) return NULL;
longHashList = calloc(sizeof(SLongHash *), maxSessions);
if (longHashList == 0) {
taosMemPoolCleanUp(longHashMemPool);
return NULL;
}
pObj = malloc(sizeof(SHashObj));
if (pObj == NULL) {
taosMemPoolCleanUp(longHashMemPool);
free(longHashList);
return NULL;
}
pObj->maxSessions = maxSessions;
pObj->longHashMemPool = longHashMemPool;
pObj->longHashList = longHashList;
pObj->dataSize = dataSize;
return pObj;
}
void sdbCloseIntHash(void *handle) {
SHashObj *pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
if (pObj->longHashMemPool) taosMemPoolCleanUp(pObj->longHashMemPool);
if (pObj->longHashList) free(pObj->longHashList);
memset(pObj, 0, sizeof(SHashObj));
free(pObj);
}
void *sdbFetchIntHashData(void *handle, void *ptr, void **ppMeta) {
SHashObj * pObj = (SHashObj *)handle;
SLongHash *pNode = (SLongHash *)ptr;
int hash = 0;
*ppMeta = NULL;
if (pObj == NULL || pObj->maxSessions <= 0) return NULL;
if (pObj->longHashList == NULL) return NULL;
if (pNode) {
hash = pNode->hash + 1;
pNode = pNode->next;
}
if (pNode == NULL) {
for (int i = hash; i < pObj->maxSessions; ++i) {
pNode = pObj->longHashList[i];
if (pNode) break;
}
}
if (pNode) *ppMeta = pNode->data;
return pNode;
}

214
src/util/src/hashstr.c
View File

@ -1,214 +0,0 @@
/*
* 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 "os.h"
#include "taosdef.h"
#define MAX_STR_LEN 40
typedef struct _str_node_t {
char string[TSDB_TABLE_ID_LEN];
int hash;
struct _str_node_t *prev;
struct _str_node_t *next;
char data[];
} SHashNode;
typedef struct {
SHashNode **hashList;
int maxSessions;
int dataSize;
} SHashObj;
int sdbHashString(void *handle, char *string) {
SHashObj * pObj = (SHashObj *)handle;
unsigned int hash = 0, hashv;
char * c;
int len = strlen(string);
c = string;
while (len >= 4) {
hash += *((int *)c);
c += 4;
len -= 4;
}
while (len > 0) {
hash += *c;
c++;
len--;
}
hashv = hash / pObj->maxSessions;
hash = (hashv + hash % pObj->maxSessions) % pObj->maxSessions;
return hash;
}
void *sdbAddStrHash(void *handle, void *key, void *pData) {
int hash;
SHashNode *pNode;
SHashObj * pObj;
char * string = (char *)key;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = sdbHashString(pObj, string);
int size = sizeof(SHashNode) + pObj->dataSize;
pNode = (SHashNode *)malloc(size);
memset(pNode, 0, size);
strcpy(pNode->string, string);
memcpy(pNode->data, pData, pObj->dataSize);
pNode->prev = 0;
pNode->next = pObj->hashList[hash];
pNode->hash = hash;
if (pObj->hashList[hash] != 0) (pObj->hashList[hash])->prev = pNode;
pObj->hashList[hash] = pNode;
return pNode->data;
}
void sdbDeleteStrHash(void *handle, void *key) {
int hash;
SHashNode *pNode;
SHashObj * pObj;
char * string = (char *)key;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
hash = sdbHashString(pObj, string);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) == 0) break;
pNode = pNode->next;
}
if (pNode) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->hashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
memset(pNode, 0, sizeof(SHashNode));
free(pNode);
}
}
void *sdbGetStrHashData(void *handle, void *key) {
int hash;
SHashNode *pNode;
SHashObj * pObj;
char * string = (char *)key;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = sdbHashString(pObj, string);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) == 0) {
break;
}
pNode = pNode->next;
}
if (pNode) return pNode->data;
return NULL;
}
void *sdbOpenStrHash(int maxSessions, int dataSize) {
SHashObj *pObj;
pObj = (SHashObj *)malloc(sizeof(SHashObj));
if (pObj == NULL) {
return NULL;
}
memset(pObj, 0, sizeof(SHashObj));
pObj->maxSessions = maxSessions;
pObj->dataSize = dataSize;
pObj->hashList = (SHashNode **)malloc(sizeof(SHashNode *) * maxSessions);
if (pObj->hashList == NULL) {
free(pObj);
return NULL;
}
memset(pObj->hashList, 0, sizeof(SHashNode *) * maxSessions);
return (void *)pObj;
}
void sdbCloseStrHash(void *handle) {
SHashObj *pObj;
SHashNode *pNode, *pNext;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
free(pNode);
pNode = pNext;
}
}
free(pObj->hashList);
}
memset(pObj, 0, sizeof(SHashObj));
free(pObj);
}
void *sdbFetchStrHashData(void *handle, void *ptr, void **ppMeta) {
SHashObj *pObj = (SHashObj *)handle;
SHashNode *pNode = (SHashNode *)ptr;
int hash = 0;
*ppMeta = NULL;
if (pObj == NULL || pObj->maxSessions <= 0) return NULL;
if (pObj->hashList == NULL) return NULL;
if (pNode) {
hash = pNode->hash + 1;
pNode = pNode->next;
}
if (pNode == NULL) {
for (int i = hash; i < pObj->maxSessions; ++i) {
pNode = pObj->hashList[i];
if (pNode) break;
}
}
if (pNode) *ppMeta = pNode->data;
return pNode;
}

290
src/util/src/ihash.c
View File

@ -1,290 +0,0 @@
/*
* 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 "os.h"
typedef struct _str_node_t {
uint64_t key;
struct _str_node_t *prev;
struct _str_node_t *next;
char data[];
} IHashNode;
typedef struct {
IHashNode **hashList;
int64_t *lockedBy;
int32_t maxSessions;
int32_t dataSize;
int32_t (*hashFp)(void *, uint64_t key);
} IHashObj;
int32_t taosHashInt(void *handle, uint64_t key) {
IHashObj *pObj = (IHashObj *)handle;
int32_t hash = key % pObj->maxSessions;
return hash;
}
static void taosLockIntHash(IHashObj *pObj, int hash);
static void taosUnlockIntHash(IHashObj *pObj, int hash);
char *taosAddIntHash(void *handle, uint64_t key, char *pData) {
int32_t hash;
IHashNode *pNode;
IHashObj * pObj;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = (*pObj->hashFp)(pObj, key);
pNode = (IHashNode *)malloc(sizeof(IHashNode) + (size_t)pObj->dataSize);
if (pNode == NULL)
return NULL;
taosLockIntHash(pObj, hash);
pNode->key = key;
if (pData != NULL) {
memcpy(pNode->data, pData, (size_t)pObj->dataSize);
}
pNode->prev = 0;
pNode->next = pObj->hashList[hash];
if (pObj->hashList[hash] != 0) (pObj->hashList[hash])->prev = pNode;
pObj->hashList[hash] = pNode;
taosUnlockIntHash(pObj, hash);
return (char *)pNode->data;
}
void taosDeleteIntHash(void *handle, uint64_t key) {
int32_t hash;
IHashNode *pNode;
IHashObj * pObj;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
hash = (*(pObj->hashFp))(pObj, key);
taosLockIntHash(pObj, hash);
pNode = pObj->hashList[hash];
while (pNode) {
if (pNode->key == key) break;
pNode = pNode->next;
}
if (pNode) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->hashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
free(pNode);
}
taosUnlockIntHash(pObj, hash);
}
char *taosGetIntHashData(void *handle, uint64_t key) {
int32_t hash;
IHashNode *pNode;
IHashObj * pObj;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
hash = (*pObj->hashFp)(pObj, key);
taosLockIntHash(pObj, hash);
pNode = pObj->hashList[hash];
while (pNode) {
if (pNode->key == key) {
break;
}
pNode = pNode->next;
}
taosUnlockIntHash(pObj, hash);
if (pNode) return pNode->data;
return NULL;
}
void *taosInitIntHash(int32_t maxSessions, int32_t dataSize, int32_t (*fp)(void *, uint64_t)) {
IHashObj *pObj;
pObj = (IHashObj *)malloc(sizeof(IHashObj));
if (pObj == NULL) {
return NULL;
}
memset(pObj, 0, sizeof(IHashObj));
pObj->maxSessions = maxSessions;
pObj->dataSize = dataSize;
pObj->hashFp = fp;
pObj->hashList = (IHashNode **)malloc(sizeof(IHashNode *) * (size_t)maxSessions);
if (pObj->hashList == NULL) {
free(pObj);
return NULL;
}
memset(pObj->hashList, 0, sizeof(IHashNode *) * (size_t)maxSessions);
pObj->lockedBy = (int64_t *)calloc(sizeof(int64_t), maxSessions);
if (pObj->lockedBy == NULL) {
free(pObj);
free(pObj->hashList);
pObj = NULL;
}
return pObj;
}
void taosCleanUpIntHash(void *handle) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
if (pObj->hashList) {
for (int32_t i = 0; i < pObj->maxSessions; ++i) {
taosLockIntHash(pObj, i);
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
free(pNode);
pNode = pNext;
}
taosUnlockIntHash(pObj, i);
}
free(pObj->hashList);
}
free(pObj->lockedBy);
free(pObj);
}
void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *)) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
taosLockIntHash(pObj, i);
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
if (fp != NULL) (*fp)(pNode->data);
free(pNode);
pNode = pNext;
}
taosUnlockIntHash(pObj, i);
}
free(pObj->hashList);
}
memset(pObj, 0, sizeof(IHashObj));
free(pObj);
}
void taosVisitIntHashWithFp(void *handle, int (*fp)(char *, void *), void *param) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
taosLockIntHash(pObj, i);
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
(*fp)(pNode->data, param);
pNode = pNext;
}
taosUnlockIntHash(pObj, i);
}
}
}
int32_t taosGetIntHashSize(void *handle) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
int32_t num = 0;
pObj = (IHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return 0;
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
taosLockIntHash(pObj, i);
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
num++;
pNode = pNext;
}
taosUnlockIntHash(pObj, i);
}
}
return num;
}
static void taosLockIntHash(IHashObj *pObj, int hash) {
int64_t tid = taosGetPthreadId();
int i = 0;
while (atomic_val_compare_exchange_64(&(pObj->lockedBy[hash]), 0, tid) != 0) {
if (++i % 1000 == 0) {
sched_yield();
}
}
}
static void taosUnlockIntHash(IHashObj *pObj, int hash) {
int64_t tid = taosGetPthreadId();
if (atomic_val_compare_exchange_64(&(pObj->lockedBy[hash]), tid, 0) != tid) {
assert(false);
}
}

310
src/util/src/shash.c
View File

@ -1,310 +0,0 @@
/*
* 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/>.
*/
#define _DEFAULT_SOURCE
#include "os.h"
#include "shash.h"
#include "tulog.h"
typedef struct _str_node_t {
char * string;
struct _str_node_t *prev;
struct _str_node_t *next;
char data[];
} SHashNode;
typedef struct {
SHashNode **hashList;
uint32_t maxSessions;
uint32_t dataSize;
uint32_t (*hashFp)(void *, char *string);
pthread_mutex_t mutex;
} SHashObj;
uint32_t taosHashString(void *handle, char *string) {
SHashObj *pObj = (SHashObj *)handle;
uint32_t hash = 0, hashv;
char * c;
c = string;
while (*c) {
hash += *((int *)c);
c += 4;
}
hashv = hash / pObj->maxSessions;
hash = (hashv + hash % pObj->maxSessions) % pObj->maxSessions;
return hash;
}
uint32_t taosHashStringStep1(void *handle, char *string) {
SHashObj *pObj = (SHashObj *)handle;
uint32_t hash = 0, hashv;
char * c;
c = string;
while (*c) {
hash += *c;
c++;
}
hashv = hash / pObj->maxSessions;
hash = (hashv + hash % pObj->maxSessions) % pObj->maxSessions;
return hash;
}
void *taosAddStrHashWithSize(void *handle, char *string, char *pData, int dataSize) {
uint32_t hash;
SHashNode *pNode;
SHashObj * pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
if (string == NULL || string[0] == 0) return NULL;
hash = (*pObj->hashFp)(pObj, string);
pthread_mutex_lock(&pObj->mutex);
pNode = (SHashNode *)malloc(sizeof(SHashNode) + (size_t)dataSize + strlen(string) + 1);
memcpy(pNode->data, pData, (size_t)dataSize);
pNode->prev = 0;
pNode->next = pObj->hashList[hash];
pNode->string = pNode->data + dataSize;
strcpy(pNode->string, string);
if (pObj->hashList[hash] != 0) (pObj->hashList[hash])->prev = pNode;
pObj->hashList[hash] = pNode;
pthread_mutex_unlock(&pObj->mutex);
uTrace("hash:%d:%s is added", hash, string);
return pNode->data;
}
void *taosAddStrHash(void *handle, char *string, char *pData) {
if (string == NULL || string[0] == 0) return NULL;
SHashObj *pObj = (SHashObj *)handle;
return taosAddStrHashWithSize(handle, string, pData, pObj->dataSize);
}
void taosDeleteStrHashNode(void *handle, char *string, void *pDeleteNode) {
uint32_t hash;
SHashNode *pNode;
SHashObj * pObj;
bool find = false;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
if (string == NULL || string[0] == 0) return;
hash = (*(pObj->hashFp))(pObj, string);
pthread_mutex_lock(&pObj->mutex);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) != 0) continue;
if (pNode->data == pDeleteNode) {
find = true;
break;
}
pNode = pNode->next;
}
if (find) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->hashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
uTrace("hash:%d:%s:%p is removed", hash, string, pNode);
free(pNode);
}
pthread_mutex_unlock(&pObj->mutex);
}
void taosDeleteStrHash(void *handle, char *string) {
uint32_t hash;
SHashNode *pNode;
SHashObj * pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return;
if (string == NULL || string[0] == 0) return;
hash = (*(pObj->hashFp))(pObj, string);
pthread_mutex_lock(&pObj->mutex);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) == 0) break;
pNode = pNode->next;
}
if (pNode) {
if (pNode->prev) {
pNode->prev->next = pNode->next;
} else {
pObj->hashList[hash] = pNode->next;
}
if (pNode->next) {
pNode->next->prev = pNode->prev;
}
uTrace("hash:%d:%s:%p is removed", hash, string, pNode);
free(pNode);
}
pthread_mutex_unlock(&pObj->mutex);
}
void *taosGetStrHashData(void *handle, char *string) {
uint32_t hash;
SHashNode *pNode;
SHashObj * pObj;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions == 0) return NULL;
if (string == NULL || string[0] == 0) return NULL;
hash = (*pObj->hashFp)(pObj, string);
pthread_mutex_lock(&pObj->mutex);
pNode = pObj->hashList[hash];
while (pNode) {
if (strcmp(pNode->string, string) == 0) {
uTrace("hash:%d:%s is retrieved", hash, string);
break;
}
pNode = pNode->next;
}
pthread_mutex_unlock(&pObj->mutex);
if (pNode) return pNode->data;
return NULL;
}
void *taosInitStrHash(uint32_t maxSessions, uint32_t dataSize, uint32_t (*fp)(void *, char *)) {
SHashObj *pObj;
pObj = (SHashObj *)malloc(sizeof(SHashObj));
if (pObj == NULL) {
return NULL;
}
memset(pObj, 0, sizeof(SHashObj));
pObj->maxSessions = maxSessions;
pObj->dataSize = dataSize;
pObj->hashFp = fp;
pObj->hashList = (SHashNode **)malloc(sizeof(SHashNode *) * (size_t)maxSessions);
if (pObj->hashList == NULL) {
free(pObj);
return NULL;
}
memset(pObj->hashList, 0, sizeof(SHashNode *) * (size_t)maxSessions);
pthread_mutex_init(&pObj->mutex, NULL);
return pObj;
}
void taosCleanUpStrHashWithFp(void *handle, void (*fp)(char *)) {
SHashObj * pObj;
SHashNode *pNode, *pNext;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return;
pthread_mutex_lock(&pObj->mutex);
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
if (fp != NULL) fp(pNode->data);
free(pNode);
pNode = pNext;
}
}
free(pObj->hashList);
}
pthread_mutex_unlock(&pObj->mutex);
pthread_mutex_destroy(&pObj->mutex);
memset(pObj, 0, sizeof(SHashObj));
free(pObj);
}
void taosCleanUpStrHash(void *handle) { taosCleanUpStrHashWithFp(handle, NULL); }
char *taosVisitStrHashWithFp(void *handle, int (*fp)(char *)) {
SHashObj * pObj;
SHashNode *pNode, *pNext;
char * pData = NULL;
pObj = (SHashObj *)handle;
if (pObj == NULL || pObj->maxSessions <= 0) return NULL;
pthread_mutex_lock(&pObj->mutex);
if (pObj->hashList) {
for (int i = 0; i < pObj->maxSessions; ++i) {
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
int flag = fp(pNode->data);
if (flag) {
pData = pNode->data;
goto VisitEnd;
}
pNode = pNext;
}
}
}
VisitEnd:
pthread_mutex_unlock(&pObj->mutex);
return pData;
}

55
src/vnode/src/vnodeMain.c
View File

@ -15,7 +15,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "ihash.h" #include "hash.h"
#include "taoserror.h" #include "taoserror.h"
#include "taosmsg.h" #include "taosmsg.h"
#include "tutil.h" #include "tutil.h"
@ -32,7 +32,6 @@
static int32_t tsOpennedVnodes; static int32_t tsOpennedVnodes;
static void *tsDnodeVnodesHash; static void *tsDnodeVnodesHash;
static void vnodeCleanUp(SVnodeObj *pVnode); static void vnodeCleanUp(SVnodeObj *pVnode);
static void vnodeBuildVloadMsg(char *pNode, void * param);
static int vnodeWalCallback(void *arg); static int vnodeWalCallback(void *arg);
static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg); static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg);
static int32_t vnodeReadCfg(SVnodeObj *pVnode); static int32_t vnodeReadCfg(SVnodeObj *pVnode);
@ -55,11 +54,11 @@ int syncGetNodesRole(tsync_h shandle, SNodesRole * cfg) { return 0; }
void syncConfirmForward(tsync_h shandle, uint64_t version, int32_t code) {} void syncConfirmForward(tsync_h shandle, uint64_t version, int32_t code) {}
#endif #endif
static void vnodeInit() { void vnodeInit() {
vnodeInitWriteFp(); vnodeInitWriteFp();
vnodeInitReadFp(); vnodeInitReadFp();
tsDnodeVnodesHash = taosInitIntHash(TSDB_MAX_VNODES, sizeof(SVnodeObj *), taosHashInt); tsDnodeVnodesHash = taosHashInit(TSDB_MAX_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true);
if (tsDnodeVnodesHash == NULL) { if (tsDnodeVnodesHash == NULL) {
vError("failed to init vnode list"); vError("failed to init vnode list");
} }
@ -69,7 +68,7 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
int32_t code; int32_t code;
pthread_once(&vnodeModuleInit, vnodeInit); pthread_once(&vnodeModuleInit, vnodeInit);
SVnodeObj *pTemp = (SVnodeObj *)taosGetIntHashData(tsDnodeVnodesHash, pVnodeCfg->cfg.vgId); SVnodeObj *pTemp = (SVnodeObj *)taosHashGet(tsDnodeVnodesHash, (const char *)&pVnodeCfg->cfg.vgId, sizeof(int32_t));
if (pTemp != NULL) { if (pTemp != NULL) {
vPrint("vgId:%d, vnode already exist, pVnode:%p", pVnodeCfg->cfg.vgId, pTemp); vPrint("vgId:%d, vnode already exist, pVnode:%p", pVnodeCfg->cfg.vgId, pTemp);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -121,7 +120,7 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
} }
int32_t vnodeDrop(int32_t vgId) { int32_t vnodeDrop(int32_t vgId) {
SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId); SVnodeObj **ppVnode = (SVnodeObj **)taosHashGet(tsDnodeVnodesHash, (const char *)&vgId, sizeof(int32_t));
if (ppVnode == NULL || *ppVnode == NULL) { if (ppVnode == NULL || *ppVnode == NULL) {
vTrace("vgId:%d, failed to drop, vgId not exist", vgId); vTrace("vgId:%d, failed to drop, vgId not exist", vgId);
return TSDB_CODE_INVALID_VGROUP_ID; return TSDB_CODE_INVALID_VGROUP_ID;
@ -148,7 +147,7 @@ int32_t vnodeAlter(void *param, SMDCreateVnodeMsg *pVnodeCfg) {
code = vnodeReadCfg(pVnode); code = vnodeReadCfg(pVnode);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
vError("vgId:%d, failed to read cfg file", pVnode->vgId); vError("vgId:%d, failed to read cfg file", pVnode->vgId);
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId); taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t));
return code; return code;
} }
@ -183,12 +182,12 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode->version = 0; pVnode->version = 0;
pVnode->tsdbCfg.tsdbId = pVnode->vgId; pVnode->tsdbCfg.tsdbId = pVnode->vgId;
pVnode->rootDir = strdup(rootDir); pVnode->rootDir = strdup(rootDir);
taosAddIntHash(tsDnodeVnodesHash, pVnode->vgId, (char *)(&pVnode)); taosHashPut(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t), (char *)(&pVnode), sizeof(SVnodeObj *));
int32_t code = vnodeReadCfg(pVnode); int32_t code = vnodeReadCfg(pVnode);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
vError("vgId:%d, failed to read cfg file", pVnode->vgId); vError("vgId:%d, failed to read cfg file", pVnode->vgId);
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId); taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t));
return code; return code;
} }
@ -213,7 +212,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode->tsdb = tsdbOpenRepo(temp, &appH); pVnode->tsdb = tsdbOpenRepo(temp, &appH);
if (pVnode->tsdb == NULL) { if (pVnode->tsdb == NULL) {
vError("vgId:%d, failed to open tsdb at %s(%s)", pVnode->vgId, temp, tstrerror(terrno)); vError("vgId:%d, failed to open tsdb at %s(%s)", pVnode->vgId, temp, tstrerror(terrno));
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId); taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t));
return terrno; return terrno;
} }
@ -249,7 +248,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
} }
int32_t vnodeClose(int32_t vgId) { int32_t vnodeClose(int32_t vgId) {
SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId); SVnodeObj **ppVnode = (SVnodeObj **)taosHashGet(tsDnodeVnodesHash, (const char *)&vgId, sizeof(int32_t));
if (ppVnode == NULL || *ppVnode == NULL) return 0; if (ppVnode == NULL || *ppVnode == NULL) return 0;
SVnodeObj *pVnode = *ppVnode; SVnodeObj *pVnode = *ppVnode;
@ -293,14 +292,14 @@ void vnodeRelease(void *pVnodeRaw) {
vTrace("vgId:%d, vnode is released, vnodes:%d", vgId, count); vTrace("vgId:%d, vnode is released, vnodes:%d", vgId, count);
if (count <= 0) { if (count <= 0) {
taosCleanUpIntHash(tsDnodeVnodesHash); taosHashCleanup(tsDnodeVnodesHash);
vnodeModuleInit = PTHREAD_ONCE_INIT; vnodeModuleInit = PTHREAD_ONCE_INIT;
tsDnodeVnodesHash = NULL; tsDnodeVnodesHash = NULL;
} }
} }
void *vnodeGetVnode(int32_t vgId) { void *vnodeGetVnode(int32_t vgId) {
SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId); SVnodeObj **ppVnode = (SVnodeObj **)taosHashGet(tsDnodeVnodesHash, (const char *)&vgId, sizeof(int32_t));
if (ppVnode == NULL || *ppVnode == NULL) { if (ppVnode == NULL || *ppVnode == NULL) {
terrno = TSDB_CODE_INVALID_VGROUP_ID; terrno = TSDB_CODE_INVALID_VGROUP_ID;
vPrint("vgId:%d, not exist", vgId); vPrint("vgId:%d, not exist", vgId);
@ -334,16 +333,8 @@ void *vnodeGetWal(void *pVnode) {
return ((SVnodeObj *)pVnode)->wal; return ((SVnodeObj *)pVnode)->wal;
} }
void vnodeBuildStatusMsg(void *param) { static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SDMStatusMsg *pStatus) {
SDMStatusMsg *pStatus = param;
taosVisitIntHashWithFp(tsDnodeVnodesHash, vnodeBuildVloadMsg, pStatus);
}
static void vnodeBuildVloadMsg(char *pNode, void * param) {
SVnodeObj *pVnode = *(SVnodeObj **) pNode;
if (pVnode->status == TAOS_VN_STATUS_DELETING) return; if (pVnode->status == TAOS_VN_STATUS_DELETING) return;
SDMStatusMsg *pStatus = param;
if (pStatus->openVnodes >= TSDB_MAX_VNODES) return; if (pStatus->openVnodes >= TSDB_MAX_VNODES) return;
SVnodeLoad *pLoad = &pStatus->load[pStatus->openVnodes++]; SVnodeLoad *pLoad = &pStatus->load[pStatus->openVnodes++];
@ -357,8 +348,24 @@ static void vnodeBuildVloadMsg(char *pNode, void * param) {
pLoad->replica = pVnode->syncCfg.replica; pLoad->replica = pVnode->syncCfg.replica;
} }
void vnodeBuildStatusMsg(void *param) {
SDMStatusMsg *pStatus = param;
SHashMutableIterator *pIter = taosHashCreateIter(tsDnodeVnodesHash);
while (taosHashIterNext(pIter)) {
SVnodeObj **pVnode = taosHashIterGet(pIter);
if (pVnode == NULL) continue;
if (*pVnode == NULL) continue;
vnodeBuildVloadMsg(*pVnode, pStatus);
pStatus++;
}
taosHashDestroyIter(pIter);
}
static void vnodeCleanUp(SVnodeObj *pVnode) { static void vnodeCleanUp(SVnodeObj *pVnode) {
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId); taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t));
if (pVnode->sync) { if (pVnode->sync) {
syncStop(pVnode->sync); syncStop(pVnode->sync);