提交 0e7ba80d 编写于 作者: S Shengliang Guan

TD-10432 vnodeint.h

上级 ad3dbbd5
...@@ -16,6 +16,7 @@ target_link_libraries( ...@@ -16,6 +16,7 @@ target_link_libraries(
PUBLIC tq PUBLIC tq
PUBLIC tsdb PUBLIC tsdb
PUBLIC wal PUBLIC wal
PUBLIC sync
PUBLIC cjson PUBLIC cjson
) )
......
...@@ -16,18 +16,19 @@ ...@@ -16,18 +16,19 @@
#ifndef _TD_VNODE_INT_H_ #ifndef _TD_VNODE_INT_H_
#define _TD_VNODE_INT_H_ #define _TD_VNODE_INT_H_
#include "os.h"
#include "amalloc.h" #include "amalloc.h"
#include "meta.h" #include "meta.h"
#include "os.h"
#include "sync.h"
#include "taosmsg.h" #include "taosmsg.h"
#include "tlog.h"
#include "tq.h" #include "tq.h"
#include "tqueue.h"
#include "trpc.h" #include "trpc.h"
#include "tsdb.h" #include "tsdb.h"
#include "tworker.h"
#include "vnode.h" #include "vnode.h"
#include "tlog.h"
#include "tqueue.h"
#include "wal.h" #include "wal.h"
#include "tworker.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
...@@ -43,51 +44,28 @@ extern int32_t vDebugFlag; ...@@ -43,51 +44,28 @@ extern int32_t vDebugFlag;
#define vTrace(...) { if (vDebugFlag & DEBUG_TRACE) { taosPrintLog("VND ", vDebugFlag, __VA_ARGS__); }} #define vTrace(...) { if (vDebugFlag & DEBUG_TRACE) { taosPrintLog("VND ", vDebugFlag, __VA_ARGS__); }}
typedef struct { typedef struct {
SMeta * pMeta; int32_t vgId; // global vnode group ID
STsdb * pTsdb; int32_t refCount; // reference count
STQ * pTQ; SMemAllocator *allocator;
SMemAllocator *allocator; SMeta *pMeta;
STsdb *pTsdb;
int32_t vgId; // global vnode group ID STQ *pTQ;
int32_t refCount; // reference count twalh pWal;
int64_t queuedWMsgSize; SyncNodeId syncNode;
int32_t queuedWMsg; taos_queue pWriteQ; // write queue
int32_t queuedRMsg; taos_queue pQueryQ; // read query queue
int32_t numOfExistQHandle; // current initialized and existed query handle in current dnode taos_queue pFetchQ; // read fetch/cancel queue
int32_t flowctrlLevel; SWalCfg walCfg;
int8_t preClose; // drop and close switch SSyncCluster syncCfg;
int8_t reserved[3]; char db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
int64_t sequence; // for topic int64_t queuedWMsgSize;
int8_t status; int32_t queuedWMsg;
int8_t role; int32_t queuedRMsg;
int8_t accessState; int32_t numOfQHandle; // current initialized and existed query handle in current dnode
int8_t isFull; int8_t status;
int8_t isCommiting; int8_t role;
int8_t dbReplica; int8_t accessState;
int8_t dropped; int8_t dropped;
int8_t dbType;
uint64_t version; // current version
uint64_t cversion; // version while commit start
uint64_t fversion; // version on saved data file
void * wqueue; // write queue
void * qqueue; // read query queue
void * fqueue; // read fetch/cancel queue
void * wal;
void * tsdb;
int64_t sync;
void * events;
void * cq; // continuous query
int32_t dbCfgVersion;
int32_t vgCfgVersion;
// STsdbCfg tsdbCfg;
#if 0
SSyncCfg syncCfg;
#endif
SWalCfg walCfg;
void * qMgmt;
char * rootDir;
tsem_t sem;
char db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
pthread_mutex_t statusMutex; pthread_mutex_t statusMutex;
} SVnode; } SVnode;
...@@ -97,6 +75,8 @@ typedef struct { ...@@ -97,6 +75,8 @@ typedef struct {
void * qhandle; // used by query and retrieve msg void * qhandle; // used by query and retrieve msg
} SVnRsp; } SVnRsp;
void vnodeSendMsgToDnode(struct SRpcEpSet *epSet, struct SRpcMsg *rpcMsg);
void vnodeSendMsgToMnode(struct SRpcMsg *rpcMsg);
void vnodeGetDnodeEp(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port); void vnodeGetDnodeEp(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port);
#ifdef __cplusplus #ifdef __cplusplus
......
...@@ -23,8 +23,8 @@ ...@@ -23,8 +23,8 @@
#include "vnodeWrite.h" #include "vnodeWrite.h"
static struct { static struct {
struct SSteps *steps; SSteps *steps;
SVnodeFp fp; SVnodeFp fp;
} tsVint; } tsVint;
int32_t vnodeInit(SVnodePara para) { int32_t vnodeInit(SVnodePara para) {
...@@ -34,11 +34,10 @@ int32_t vnodeInit(SVnodePara para) { ...@@ -34,11 +34,10 @@ int32_t vnodeInit(SVnodePara para) {
if (steps == NULL) return -1; if (steps == NULL) return -1;
taosStepAdd(steps, "vnode-main", vnodeInitMain, vnodeCleanupMain); taosStepAdd(steps, "vnode-main", vnodeInitMain, vnodeCleanupMain);
taosStepAdd(steps, "vnode-worker",vnodeInitWorker, vnodeCleanupWorker); taosStepAdd(steps, "vnode-worker", vnodeInitWorker, vnodeCleanupWorker);
taosStepAdd(steps, "vnode-read", vnodeInitRead, vnodeCleanupRead); taosStepAdd(steps, "vnode-read", vnodeInitRead, vnodeCleanupRead);
taosStepAdd(steps, "vnode-mgmt", vnodeInitMgmt, vnodeCleanupMgmt); taosStepAdd(steps, "vnode-mgmt", vnodeInitMgmt, vnodeCleanupMgmt);
taosStepAdd(steps, "vnode-write", vnodeInitWrite, vnodeCleanupWrite); taosStepAdd(steps, "vnode-write", vnodeInitWrite, vnodeCleanupWrite);
// taosStepAdd(steps, "vnode-queue", tsdbInitCommitQueue, tsdbDestroyCommitQueue);
tsVint.steps = steps; tsVint.steps = steps;
return taosStepExec(tsVint.steps); return taosStepExec(tsVint.steps);
...@@ -48,4 +47,10 @@ void vnodeCleanup() { taosStepCleanup(tsVint.steps); } ...@@ -48,4 +47,10 @@ void vnodeCleanup() { taosStepCleanup(tsVint.steps); }
void vnodeGetDnodeEp(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port) { void vnodeGetDnodeEp(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port) {
return (*tsVint.fp.GetDnodeEp)(dnodeId, ep, fqdn, port); return (*tsVint.fp.GetDnodeEp)(dnodeId, ep, fqdn, port);
} }
\ No newline at end of file
void vnodeSendMsgToDnode(struct SRpcEpSet *epSet, struct SRpcMsg *rpcMsg) {
(*tsVint.fp.SendMsgToDnode)(epSet, rpcMsg);
}
void vnodeSendMsgToMnode(struct SRpcMsg *rpcMsg) { return (*tsVint.fp.SendMsgToMnode)(rpcMsg); }
...@@ -75,15 +75,16 @@ SVnode *vnodeAcquire(int32_t vgId) { ...@@ -75,15 +75,16 @@ SVnode *vnodeAcquire(int32_t vgId) {
} }
SVnode *vnodeAcquireNotClose(int32_t vgId) { SVnode *vnodeAcquireNotClose(int32_t vgId) {
SVnode *pVnode = vnodeAcquire(vgId); // SVnode *pVnode = vnodeAcquire(vgId);
if (pVnode != NULL && pVnode->preClose == 1) { // if (pVnode != NULL && pVnode->preClose == 1) {
vnodeRelease(pVnode); // vnodeRelease(pVnode);
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID; // terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
vDebug("vgId:%d, not exist, pre closing", vgId); // vDebug("vgId:%d, not exist, pre closing", vgId);
return NULL; // return NULL;
} // }
return pVnode; // return pVnode;
return NULL;
} }
void vnodeRelease(SVnode *pVnode) { void vnodeRelease(SVnode *pVnode) {
...@@ -287,6 +288,7 @@ static int32_t vnodeAlterImp(SVnode *pVnode, SCreateVnodeMsg *pVnodeCfg) { ...@@ -287,6 +288,7 @@ static int32_t vnodeAlterImp(SVnode *pVnode, SCreateVnodeMsg *pVnodeCfg) {
} }
int32_t vnodeAlter(SVnode *pVnode, SCreateVnodeMsg *pVnodeCfg) { int32_t vnodeAlter(SVnode *pVnode, SCreateVnodeMsg *pVnodeCfg) {
#if 0
vDebug("vgId:%d, current dbCfgVersion:%d vgCfgVersion:%d, input dbCfgVersion:%d vgCfgVersion:%d", pVnode->vgId, vDebug("vgId:%d, current dbCfgVersion:%d vgCfgVersion:%d, input dbCfgVersion:%d vgCfgVersion:%d", pVnode->vgId,
pVnode->dbCfgVersion, pVnode->vgCfgVersion, pVnodeCfg->cfg.dbCfgVersion, pVnodeCfg->cfg.vgCfgVersion); pVnode->dbCfgVersion, pVnode->vgCfgVersion, pVnodeCfg->cfg.dbCfgVersion, pVnodeCfg->cfg.vgCfgVersion);
...@@ -304,6 +306,8 @@ int32_t vnodeAlter(SVnode *pVnode, SCreateVnodeMsg *pVnodeCfg) { ...@@ -304,6 +306,8 @@ int32_t vnodeAlter(SVnode *pVnode, SCreateVnodeMsg *pVnodeCfg) {
} }
return code; return code;
#endif
return 0;
} }
static void vnodeFindWalRootDir(int32_t vgId, char *walRootDir) { static void vnodeFindWalRootDir(int32_t vgId, char *walRootDir) {
...@@ -371,10 +375,10 @@ int32_t vnodeOpen(int32_t vgId) { ...@@ -371,10 +375,10 @@ int32_t vnodeOpen(int32_t vgId) {
pVnode->fversion = pVnode->version; pVnode->fversion = pVnode->version;
pVnode->wqueue = vnodeAllocWriteQueue(pVnode); pVnode->pWriteQ = vnodeAllocWriteQueue(pVnode);
pVnode->qqueue = vnodeAllocQueryQueue(pVnode); pVnode->pQueryQ = vnodeAllocQueryQueue(pVnode);
pVnode->fqueue = vnodeAllocFetchQueue(pVnode); pVnode->pFetchQ = vnodeAllocFetchQueue(pVnode);
if (pVnode->wqueue == NULL || pVnode->qqueue == NULL || pVnode->fqueue == NULL) { if (pVnode->pWriteQ == NULL || pVnode->pQueryQ == NULL || pVnode->pFetchQ == NULL) {
vnodeCleanUp(pVnode); vnodeCleanUp(pVnode);
return terrno; return terrno;
} }
...@@ -467,7 +471,7 @@ int32_t vnodeClose(int32_t vgId) { ...@@ -467,7 +471,7 @@ int32_t vnodeClose(int32_t vgId) {
return 0; return 0;
} }
pVnode->preClose = 1; // pVnode->preClose = 1;
vDebug("vgId:%d, vnode will be closed, pVnode:%p", pVnode->vgId, pVnode); vDebug("vgId:%d, vnode will be closed, pVnode:%p", pVnode->vgId, pVnode);
vnodeRelease(pVnode); vnodeRelease(pVnode);
...@@ -516,19 +520,19 @@ void vnodeDestroy(SVnode *pVnode) { ...@@ -516,19 +520,19 @@ void vnodeDestroy(SVnode *pVnode) {
pVnode->wal = NULL; pVnode->wal = NULL;
} }
if (pVnode->wqueue) { if (pVnode->pWriteQ) {
vnodeFreeWriteQueue(pVnode->wqueue); vnodeFreeWriteQueue(pVnode->pWriteQ);
pVnode->wqueue = NULL; pVnode->pWriteQ = NULL;
} }
if (pVnode->qqueue) { if (pVnode->pQueryQ) {
vnodeFreeQueryQueue(pVnode->qqueue); vnodeFreeQueryQueue(pVnode->pQueryQ);
pVnode->qqueue = NULL; pVnode->pQueryQ = NULL;
} }
if (pVnode->fqueue) { if (pVnode->pFetchQ) {
vnodeFreeFetchQueue(pVnode->fqueue); vnodeFreeFetchQueue(pVnode->pFetchQ);
pVnode->fqueue = NULL; pVnode->pFetchQ = NULL;
} }
tfree(pVnode->rootDir); tfree(pVnode->rootDir);
......
...@@ -81,9 +81,9 @@ static int32_t vnodeWriteToRQueue(SVnode *pVnode, void *pCont, int32_t contLen, ...@@ -81,9 +81,9 @@ static int32_t vnodeWriteToRQueue(SVnode *pVnode, void *pCont, int32_t contLen,
atomic_add_fetch_32(&pVnode->queuedRMsg, 1); atomic_add_fetch_32(&pVnode->queuedRMsg, 1);
if (pRead->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pRead->msgType == TSDB_MSG_TYPE_FETCH) { if (pRead->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pRead->msgType == TSDB_MSG_TYPE_FETCH) {
return taosWriteQitem(pVnode->fqueue, qtype, pRead); return taosWriteQitem(pVnode->pFetchQ, qtype, pRead);
} else { } else {
return taosWriteQitem(pVnode->qqueue, qtype, pRead); return taosWriteQitem(pVnode->pQueryQ, qtype, pRead);
} }
} }
......
...@@ -158,7 +158,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SReadMsg *pRead) { ...@@ -158,7 +158,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SReadMsg *pRead) {
} }
} }
int32_t remain = atomic_add_fetch_32(&pVnode->numOfExistQHandle, 1); int32_t remain = atomic_add_fetch_32(&pVnode->numOfQHandle, 1);
vTrace("vgId:%d, new qhandle created, total qhandle:%d", pVnode->vgId, remain); vTrace("vgId:%d, new qhandle created, total qhandle:%d", pVnode->vgId, remain);
} else { } else {
assert(pCont != NULL); assert(pCont != NULL);
...@@ -203,7 +203,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SReadMsg *pRead) { ...@@ -203,7 +203,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SReadMsg *pRead) {
// If the building of result is not required, simply free it. Otherwise, mandatorily free the qhandle // If the building of result is not required, simply free it. Otherwise, mandatorily free the qhandle
if (freehandle || (!buildRes)) { if (freehandle || (!buildRes)) {
if (freehandle) { if (freehandle) {
int32_t remain = atomic_sub_fetch_32(&pVnode->numOfExistQHandle, 1); int32_t remain = atomic_sub_fetch_32(&pVnode->numOfQHandle, 1);
vTrace("vgId:%d, QInfo:%p, start to free qhandle, remain qhandle:%d", pVnode->vgId, *qhandle, remain); vTrace("vgId:%d, QInfo:%p, start to free qhandle, remain qhandle:%d", pVnode->vgId, *qhandle, remain);
} }
...@@ -282,7 +282,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SReadMsg *pRead) { ...@@ -282,7 +282,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SReadMsg *pRead) {
// kill current query and free corresponding resources. // kill current query and free corresponding resources.
if (pRetrieve->free == 1) { if (pRetrieve->free == 1) {
int32_t remain = atomic_sub_fetch_32(&pVnode->numOfExistQHandle, 1); int32_t remain = atomic_sub_fetch_32(&pVnode->numOfQHandle, 1);
vWarn("vgId:%d, QInfo:%" PRIx64 "-%p, retrieve msg received to kill query and free qhandle, remain qhandle:%d", vWarn("vgId:%d, QInfo:%" PRIx64 "-%p, retrieve msg received to kill query and free qhandle, remain qhandle:%d",
pVnode->vgId, pRetrieve->qId, *handle, remain); pVnode->vgId, pRetrieve->qId, *handle, remain);
...@@ -296,7 +296,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SReadMsg *pRead) { ...@@ -296,7 +296,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SReadMsg *pRead) {
// register the qhandle to connect to quit query immediate if connection is broken // register the qhandle to connect to quit query immediate if connection is broken
if (vnodeNotifyCurrentQhandle(pRead->rpcHandle, pRetrieve->qId, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) { if (vnodeNotifyCurrentQhandle(pRead->rpcHandle, pRetrieve->qId, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
int32_t remain = atomic_sub_fetch_32(&pVnode->numOfExistQHandle, 1); int32_t remain = atomic_sub_fetch_32(&pVnode->numOfQHandle, 1);
vError("vgId:%d, QInfo:%" PRIu64 "-%p, retrieve discarded since link is broken, conn:%p, remain qhandle:%d", vError("vgId:%d, QInfo:%" PRIu64 "-%p, retrieve discarded since link is broken, conn:%p, remain qhandle:%d",
pVnode->vgId, pRetrieve->qhandle, *handle, pRead->rpcHandle, remain); pVnode->vgId, pRetrieve->qhandle, *handle, pRead->rpcHandle, remain);
...@@ -333,7 +333,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SReadMsg *pRead) { ...@@ -333,7 +333,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SReadMsg *pRead) {
// If qhandle is not added into vread queue, the query should be completed already or paused with error. // If qhandle is not added into vread queue, the query should be completed already or paused with error.
// Here free qhandle immediately // Here free qhandle immediately
if (freeHandle) { if (freeHandle) {
int32_t remain = atomic_sub_fetch_32(&pVnode->numOfExistQHandle, 1); int32_t remain = atomic_sub_fetch_32(&pVnode->numOfQHandle, 1);
vTrace("vgId:%d, QInfo:%p, start to free qhandle, remain qhandle:%d", pVnode->vgId, *handle, remain); vTrace("vgId:%d, QInfo:%p, start to free qhandle, remain qhandle:%d", pVnode->vgId, *handle, remain);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true); qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
} }
......
...@@ -58,10 +58,12 @@ int32_t vnodeReadVersion(SVnode *pVnode) { ...@@ -58,10 +58,12 @@ int32_t vnodeReadVersion(SVnode *pVnode) {
vError("vgId:%d, failed to read %s, version not found", pVnode->vgId, file); vError("vgId:%d, failed to read %s, version not found", pVnode->vgId, file);
goto PARSE_VER_ERROR; goto PARSE_VER_ERROR;
} }
#if 0
pVnode->version = (uint64_t)ver->valueint; pVnode->version = (uint64_t)ver->valueint;
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
vInfo("vgId:%d, read %s successfully, fver:%" PRIu64, pVnode->vgId, file, pVnode->version); vInfo("vgId:%d, read %s successfully, fver:%" PRIu64, pVnode->vgId, file, pVnode->version);
#endif
PARSE_VER_ERROR: PARSE_VER_ERROR:
if (content != NULL) free(content); if (content != NULL) free(content);
...@@ -85,16 +87,17 @@ int32_t vnodeSaveVersion(SVnode *pVnode) { ...@@ -85,16 +87,17 @@ int32_t vnodeSaveVersion(SVnode *pVnode) {
int32_t maxLen = 100; int32_t maxLen = 100;
char * content = calloc(1, maxLen + 1); char * content = calloc(1, maxLen + 1);
#if 0
len += snprintf(content + len, maxLen - len, "{\n"); len += snprintf(content + len, maxLen - len, "{\n");
len += snprintf(content + len, maxLen - len, " \"version\": %" PRIu64 "\n", pVnode->fversion); len += snprintf(content + len, maxLen - len, " \"version\": %" PRIu64 "\n", pVnode->fversion);
len += snprintf(content + len, maxLen - len, "}\n"); len += snprintf(content + len, maxLen - len, "}\n");
#endif
fwrite(content, 1, len, fp); fwrite(content, 1, len, fp);
taosFsyncFile(fileno(fp)); taosFsyncFile(fileno(fp));
fclose(fp); fclose(fp);
free(content); free(content);
terrno = 0; terrno = 0;
vInfo("vgId:%d, successed to write %s, fver:%" PRIu64, pVnode->vgId, file, pVnode->fversion); // vInfo("vgId:%d, successed to write %s, fver:%" PRIu64, pVnode->vgId, file, pVnode->fversion);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
\ No newline at end of file
...@@ -96,7 +96,7 @@ static int32_t vnodeWriteToWQueue(SVnode *pVnode, SWalHead *pHead, int32_t qtype ...@@ -96,7 +96,7 @@ static int32_t vnodeWriteToWQueue(SVnode *pVnode, SWalHead *pHead, int32_t qtype
atomic_add_fetch_32(&tsVwrite.queuedMsgs, 1); atomic_add_fetch_32(&tsVwrite.queuedMsgs, 1);
atomic_add_fetch_32(&pVnode->refCount, 1); atomic_add_fetch_32(&pVnode->refCount, 1);
atomic_add_fetch_32(&pVnode->queuedWMsg, 1); atomic_add_fetch_32(&pVnode->queuedWMsg, 1);
taosWriteQitem(pVnode->wqueue, pWrite->qtype, pWrite); taosWriteQitem(pVnode->pWriteQ, pWrite->qtype, pWrite);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -153,10 +153,10 @@ static bool vnodeProcessWriteStart(SVnode *pVnode, SVnWriteMsg *pWrite, int32_t ...@@ -153,10 +153,10 @@ static bool vnodeProcessWriteStart(SVnode *pVnode, SVnWriteMsg *pWrite, int32_t
#if 0 #if 0
pWrite->code = walWrite(pVnode->wal, pHead); pWrite->code = walWrite(pVnode->wal, pHead);
if (pWrite->code < 0) return false; if (pWrite->code < 0) return false;
#endif
pVnode->version = pHead->version;
pVnode->version = pHead->version;
#endif
// write data locally // write data locally
switch (msgType) { switch (msgType) {
case TSDB_MSG_TYPE_SUBMIT: case TSDB_MSG_TYPE_SUBMIT:
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册