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

refact mnode queue

This commit is contained in:
Shengliang Guan 2021-12-28 01:38:15 -08:00
parent d60941d4e3
commit 041e671601
13 changed files with 160 additions and 367 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
include/dnode/mnode/mnode.h
View File

@ -147,28 +147,12 @@ void mndCleanupMsg(SMnodeMsg *pMsg);
void mndSendRsp(SMnodeMsg *pMsg, int32_t code);
/**
* @brief Process the read request.
* @brief Process the read, write, sync request.
*
* @param pMsg The request msg.
* @return int32_t 0 for success, -1 for failure.
*/
void mndProcessReadMsg(SMnodeMsg *pMsg);
/**
* @brief Process the write request.
*
* @param pMsg The request msg.
* @return int32_t 0 for success, -1 for failure.
*/
void mndProcessWriteMsg(SMnodeMsg *pMsg);
/**
* @brief Process the sync request.
*
* @param pMsg The request msg.
* @return int32_t 0 for success, -1 for failure.
*/
void mndProcessSyncMsg(SMnodeMsg *pMsg);
void mndProcessMsg(SMnodeMsg *pMsg);
#ifdef __cplusplus
}

2
include/dnode/snode/snode.h
View File

@ -79,7 +79,7 @@ int32_t sndGetLoad(SSnode *pSnode, SSnodeLoad *pLoad);
* @param pRsp The response message
* @return int32_t 0 for success, -1 for failure
*/
int32_t sndProcessWriteMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
int32_t sndProcessMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
/**
* @brief Drop a snode.

32
source/dnode/mgmt/impl/inc/dndInt.h
View File

@ -65,8 +65,10 @@ typedef struct {
void *queueFp;
SDnode *pDnode;
taos_queue queue;
SWorkerPool pool;
SMWorkerPool mpool;
union {
SWorkerPool pool;
SMWorkerPool mpool;
};
} SDnodeWorker;
typedef struct {
@ -95,21 +97,17 @@ typedef struct {
} SDnodeMgmt;
typedef struct {
int32_t refCount;
int8_t deployed;
int8_t dropped;
int8_t replica;
int8_t selfIndex;
SReplica replicas[TSDB_MAX_REPLICA];
char *file;
SMnode *pMnode;
SRWLatch latch;
taos_queue pReadQ;
taos_queue pWriteQ;
taos_queue pSyncQ;
SWorkerPool readPool;
SWorkerPool writePool;
SWorkerPool syncPool;
int32_t refCount;
int8_t deployed;
int8_t dropped;
SMnode *pMnode;
SRWLatch latch;
SDnodeWorker readWorker;
SDnodeWorker writeWorker;
SDnodeWorker syncWorker;
int8_t replica;
int8_t selfIndex;
SReplica replicas[TSDB_MAX_REPLICA];
} SMnodeMgmt;
typedef struct {

22
source/dnode/mgmt/impl/src/dndBnode.c
View File

@ -140,20 +140,22 @@ static int32_t dndWriteBnodeFile(SDnode *pDnode) {
fclose(fp);
free(content);
if (taosRenameFile(file, file) != 0) {
char realfile[PATH_MAX + 20];
snprintf(realfile, PATH_MAX + 20, "%s/bnode.json", pDnode->dir.dnode);
if (taosRenameFile(file, realfile) != 0) {
terrno = TSDB_CODE_DND_BNODE_WRITE_FILE_ERROR;
dError("failed to rename %s since %s", file, terrstr());
return -1;
}
dInfo("successed to write %s, deployed:%d dropped:%d", file, pMgmt->deployed, pMgmt->dropped);
dInfo("successed to write %s, deployed:%d dropped:%d", realfile, pMgmt->deployed, pMgmt->dropped);
return 0;
}
static int32_t dndStartBnodeWorker(SDnode *pDnode) {
SBnodeMgmt *pMgmt = &pDnode->bmgmt;
if (dndInitWorker(pDnode, &pMgmt->writeWorker, DND_WORKER_SINGLE, "bnode-write", 0, 1,
(FProcessItem)dndProcessBnodeQueue) != 0) {
if (dndInitWorker(pDnode, &pMgmt->writeWorker, DND_WORKER_MULTI, "bnode-write", 0, 1, dndProcessBnodeQueue) != 0) {
dError("failed to start bnode write worker since %s", terrstr());
return -1;
}
@ -202,7 +204,9 @@ static int32_t dndOpenBnode(SDnode *pDnode) {
return -1;
}
pMgmt->deployed = 1;
if (dndWriteBnodeFile(pDnode) != 0) {
pMgmt->deployed = 0;
dError("failed to write bnode file since %s", terrstr());
dndStopBnodeWorker(pDnode);
bndClose(pBnode);
@ -211,7 +215,6 @@ static int32_t dndOpenBnode(SDnode *pDnode) {
taosWLockLatch(&pMgmt->latch);
pMgmt->pBnode = pBnode;
pMgmt->deployed = 1;
taosWUnLockLatch(&pMgmt->latch);
dInfo("bnode open successfully");
@ -243,6 +246,8 @@ static int32_t dndDropBnode(SDnode *pDnode) {
dndReleaseBnode(pDnode, pBnode);
dndStopBnodeWorker(pDnode);
pMgmt->deployed = 0;
dndWriteBnodeFile(pDnode);
bndClose(pBnode);
pMgmt->pBnode = NULL;
bndDestroy(pDnode->dir.bnode);
@ -353,7 +358,12 @@ int32_t dndInitBnode(SDnode *pDnode) {
return -1;
}
if (pMgmt->dropped) return 0;
if (pMgmt->dropped) {
dInfo("bnode has been deployed and needs to be deleted");
bndDestroy(pDnode->dir.bnode);
return 0;
}
if (!pMgmt->deployed) return 0;
return dndOpenBnode(pDnode);

368
source/dnode/mgmt/impl/src/dndMnode.c
View File

@ -17,42 +17,9 @@
#include "dndMnode.h"
#include "dndDnode.h"
#include "dndTransport.h"
#include "dndWorker.h"
static int32_t dndInitMnodeReadWorker(SDnode *pDnode);
static int32_t dndInitMnodeWriteWorker(SDnode *pDnode);
static int32_t dndInitMnodeSyncWorker(SDnode *pDnode);
static void dndCleanupMnodeReadWorker(SDnode *pDnode);
static void dndCleanupMnodeWriteWorker(SDnode *pDnode);
static void dndCleanupMnodeSyncWorker(SDnode *pDnode);
static void dndCleanupMnodeMgmtWorker(SDnode *pDnode);
static int32_t dndAllocMnodeReadQueue(SDnode *pDnode);
static int32_t dndAllocMnodeWriteQueue(SDnode *pDnode);
static int32_t dndAllocMnodeSyncQueue(SDnode *pDnode);
static void dndFreeMnodeReadQueue(SDnode *pDnode);
static void dndFreeMnodeWriteQueue(SDnode *pDnode);
static void dndFreeMnodeSyncQueue(SDnode *pDnode);
static void dndFreeMnodeMgmtQueue(SDnode *pDnode);
static void dndProcessMnodeReadQueue(SDnode *pDnode, SMnodeMsg *pMsg);
static void dndProcessMnodeWriteQueue(SDnode *pDnode, SMnodeMsg *pMsg);
static void dndProcessMnodeSyncQueue(SDnode *pDnode, SMnodeMsg *pMsg);
static int32_t dndWriteMnodeMsgToQueue(SMnode *pMnode, taos_queue pQueue, SRpcMsg *pRpcMsg);
void dndProcessMnodeReadMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
void dndProcessMnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
void dndProcessMnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
static int32_t dndStartMnodeWorker(SDnode *pDnode);
static void dndStopMnodeWorker(SDnode *pDnode);
static SMnode *dndAcquireMnode(SDnode *pDnode);
static void dndReleaseMnode(SDnode *pDnode, SMnode *pMnode);
static int32_t dndReadMnodeFile(SDnode *pDnode);
static int32_t dndWriteMnodeFile(SDnode *pDnode);
static int32_t dndOpenMnode(SDnode *pDnode, SMnodeOpt *pOption);
static int32_t dndAlterMnode(SDnode *pDnode, SMnodeOpt *pOption);
static int32_t dndDropMnode(SDnode *pDnode);
static void dndProcessMnodeQueue(SDnode *pDnode, SMnodeMsg *pMsg);
static SMnode *dndAcquireMnode(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
@ -97,49 +64,52 @@ static int32_t dndReadMnodeFile(SDnode *pDnode) {
char *content = calloc(1, maxLen + 1);
cJSON *root = NULL;
FILE *fp = fopen(pMgmt->file, "r");
char file[PATH_MAX + 20];
snprintf(file, PATH_MAX + 20, "%s/mnode.json", pDnode->dir.dnode);
FILE *fp = fopen(file, "r");
if (fp == NULL) {
dDebug("file %s not exist", pMgmt->file);
dDebug("file %s not exist", file);
code = 0;
goto PRASE_MNODE_OVER;
}
len = (int32_t)fread(content, 1, maxLen, fp);
if (len <= 0) {
dError("failed to read %s since content is null", pMgmt->file);
dError("failed to read %s since content is null", file);
goto PRASE_MNODE_OVER;
}
content[len] = 0;
root = cJSON_Parse(content);
if (root == NULL) {
dError("failed to read %s since invalid json format", pMgmt->file);
dError("failed to read %s since invalid json format", file);
goto PRASE_MNODE_OVER;
}
cJSON *deployed = cJSON_GetObjectItem(root, "deployed");
if (!deployed || deployed->type != cJSON_Number) {
dError("failed to read %s since deployed not found", pMgmt->file);
dError("failed to read %s since deployed not found", file);
goto PRASE_MNODE_OVER;
}
pMgmt->deployed = deployed->valueint;
cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
if (!dropped || dropped->type != cJSON_Number) {
dError("failed to read %s since dropped not found", pMgmt->file);
dError("failed to read %s since dropped not found", file);
goto PRASE_MNODE_OVER;
}
pMgmt->dropped = dropped->valueint;
cJSON *mnodes = cJSON_GetObjectItem(root, "mnodes");
if (!mnodes || mnodes->type != cJSON_Array) {
dError("failed to read %s since nodes not found", pMgmt->file);
dError("failed to read %s since nodes not found", file);
goto PRASE_MNODE_OVER;
}
pMgmt->replica = cJSON_GetArraySize(mnodes);
if (pMgmt->replica <= 0 || pMgmt->replica > TSDB_MAX_REPLICA) {
dError("failed to read %s since mnodes size %d invalid", pMgmt->file, pMgmt->replica);
dError("failed to read %s since mnodes size %d invalid", file, pMgmt->replica);
goto PRASE_MNODE_OVER;
}
@ -151,28 +121,28 @@ static int32_t dndReadMnodeFile(SDnode *pDnode) {
cJSON *id = cJSON_GetObjectItem(node, "id");
if (!id || id->type != cJSON_Number) {
dError("failed to read %s since id not found", pMgmt->file);
dError("failed to read %s since id not found", file);
goto PRASE_MNODE_OVER;
}
pReplica->id = id->valueint;
cJSON *fqdn = cJSON_GetObjectItem(node, "fqdn");
if (!fqdn || fqdn->type != cJSON_String || fqdn->valuestring == NULL) {
dError("failed to read %s since fqdn not found", pMgmt->file);
dError("failed to read %s since fqdn not found", file);
goto PRASE_MNODE_OVER;
}
tstrncpy(pReplica->fqdn, fqdn->valuestring, TSDB_FQDN_LEN);
cJSON *port = cJSON_GetObjectItem(node, "port");
if (!port || port->type != cJSON_Number) {
dError("failed to read %s since port not found", pMgmt->file);
dError("failed to read %s since port not found", file);
goto PRASE_MNODE_OVER;
}
pReplica->port = port->valueint;
}
code = 0;
dDebug("succcessed to read file %s, deployed:%d dropped:%d", pMgmt->file, pMgmt->deployed, pMgmt->dropped);
dDebug("succcessed to read file %s, deployed:%d dropped:%d", file, pMgmt->deployed, pMgmt->dropped);
PRASE_MNODE_OVER:
if (content != NULL) free(content);
@ -186,8 +156,8 @@ PRASE_MNODE_OVER:
static int32_t dndWriteMnodeFile(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
char file[PATH_MAX + 20] = {0};
snprintf(file, sizeof(file), "%s.bak", pMgmt->file);
char file[PATH_MAX + 20];
snprintf(file, PATH_MAX + 20, "%s/mnode.json.bak", pDnode->dir.dnode);
FILE *fp = fopen(file, "w");
if (fp == NULL) {
@ -223,47 +193,36 @@ static int32_t dndWriteMnodeFile(SDnode *pDnode) {
fclose(fp);
free(content);
if (taosRenameFile(file, pMgmt->file) != 0) {
char realfile[PATH_MAX + 20];
snprintf(realfile, PATH_MAX + 20, "%s/mnode.json", pDnode->dir.dnode);
if (taosRenameFile(file, realfile) != 0) {
terrno = TSDB_CODE_DND_MNODE_WRITE_FILE_ERROR;
dError("failed to rename %s since %s", pMgmt->file, terrstr());
dError("failed to rename %s since %s", file, terrstr());
return -1;
}
dInfo("successed to write %s, deployed:%d dropped:%d", pMgmt->file, pMgmt->deployed, pMgmt->dropped);
dInfo("successed to write %s, deployed:%d dropped:%d", realfile, pMgmt->deployed, pMgmt->dropped);
return 0;
}
static int32_t dndStartMnodeWorker(SDnode *pDnode) {
if (dndInitMnodeReadWorker(pDnode) != 0) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
if (dndInitWorker(pDnode, &pMgmt->readWorker, DND_WORKER_SINGLE, "mnode-read", 0, 1, dndProcessMnodeQueue) != 0) {
dError("failed to start mnode read worker since %s", terrstr());
return -1;
}
if (dndInitMnodeWriteWorker(pDnode) != 0) {
if (dndInitWorker(pDnode, &pMgmt->writeWorker, DND_WORKER_SINGLE, "mnode-write", 0, 1, dndProcessMnodeQueue) != 0) {
dError("failed to start mnode write worker since %s", terrstr());
return -1;
}
if (dndInitMnodeSyncWorker(pDnode) != 0) {
if (dndInitWorker(pDnode, &pMgmt->syncWorker, DND_WORKER_SINGLE, "mnode-sync", 0, 1, dndProcessMnodeQueue) != 0) {
dError("failed to start mnode sync worker since %s", terrstr());
return -1;
}
if (dndAllocMnodeReadQueue(pDnode) != 0) {
dError("failed to alloc mnode read queue since %s", terrstr());
return -1;
}
if (dndAllocMnodeWriteQueue(pDnode) != 0) {
dError("failed to alloc mnode write queue since %s", terrstr());
return -1;
}
if (dndAllocMnodeSyncQueue(pDnode) != 0) {
dError("failed to alloc mnode sync queue since %s", terrstr());
return -1;
}
return 0;
}
@ -274,18 +233,13 @@ static void dndStopMnodeWorker(SDnode *pDnode) {
pMgmt->deployed = 0;
taosWUnLockLatch(&pMgmt->latch);
while (pMgmt->refCount > 1) taosMsleep(10);
while (!taosQueueEmpty(pMgmt->pReadQ)) taosMsleep(10);
while (!taosQueueEmpty(pMgmt->pWriteQ)) taosMsleep(10);
while (!taosQueueEmpty(pMgmt->pSyncQ)) taosMsleep(10);
while (pMgmt->refCount > 1) {
taosMsleep(10);
}
dndCleanupMnodeReadWorker(pDnode);
dndCleanupMnodeWriteWorker(pDnode);
dndCleanupMnodeSyncWorker(pDnode);
dndFreeMnodeReadQueue(pDnode);
dndFreeMnodeWriteQueue(pDnode);
dndFreeMnodeSyncQueue(pDnode);
dndCleanupWorker(&pMgmt->readWorker);
dndCleanupWorker(&pMgmt->writeWorker);
dndCleanupWorker(&pMgmt->syncWorker);
}
static bool dndNeedDeployMnode(SDnode *pDnode) {
@ -383,28 +337,21 @@ static int32_t dndOpenMnode(SDnode *pDnode, SMnodeOpt *pOption) {
dError("failed to open mnode since %s", terrstr());
return -1;
}
pMgmt->deployed = 1;
int32_t code = dndWriteMnodeFile(pDnode);
if (code != 0) {
dError("failed to write mnode file since %s", terrstr());
code = terrno;
pMgmt->deployed = 0;
if (dndStartMnodeWorker(pDnode) != 0) {
dError("failed to start mnode worker since %s", terrstr());
mndClose(pMnode);
mndDestroy(pDnode->dir.mnode);
terrno = code;
return -1;
}
code = dndStartMnodeWorker(pDnode);
if (code != 0) {
dError("failed to start mnode worker since %s", terrstr());
code = terrno;
pMgmt->deployed = 1;
if (dndWriteMnodeFile(pDnode) != 0) {
dError("failed to write mnode file since %s", terrstr());
pMgmt->deployed = 0;
dndStopMnodeWorker(pDnode);
mndClose(pMnode);
mndDestroy(pDnode->dir.mnode);
terrno = code;
return -1;
}
@ -461,6 +408,7 @@ static int32_t dndDropMnode(SDnode *pDnode) {
dndReleaseMnode(pDnode, pMnode);
dndStopMnodeWorker(pDnode);
pMgmt->deployed = 0;
dndWriteMnodeFile(pDnode);
mndClose(pMnode);
pMgmt->pMnode = NULL;
@ -528,13 +476,12 @@ int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
}
}
static void dndProcessMnodeReadQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
static void dndProcessMnodeQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) {
mndProcessReadMsg(pMsg);
mndProcessMsg(pMsg);
dndReleaseMnode(pDnode, pMnode);
} else {
mndSendRsp(pMsg, terrno);
@ -543,208 +490,43 @@ static void dndProcessMnodeReadQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
mndCleanupMsg(pMsg);
}
static void dndProcessMnodeWriteQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
static void dndWriteMnodeMsgToWorker(SDnode *pDnode, SDnodeWorker *pWorker, SRpcMsg *pRpcMsg) {
int32_t code = TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) {
mndProcessWriteMsg(pMsg);
dndReleaseMnode(pDnode, pMnode);
} else {
mndSendRsp(pMsg, terrno);
SMnodeMsg *pMsg = mndInitMsg(pMnode, pRpcMsg);
if (pMsg == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
} else {
code = dndWriteMsgToWorker(pWorker, pMsg, 0);
}
if (code != 0) {
mndCleanupMsg(pMsg);
}
}
dndReleaseMnode(pDnode, pMnode);
mndCleanupMsg(pMsg);
}
static void dndProcessMnodeSyncQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) {
mndProcessSyncMsg(pMsg);
dndReleaseMnode(pDnode, pMnode);
} else {
mndSendRsp(pMsg, terrno);
if (code != 0) {
if (pRpcMsg->msgType & 1u) {
SRpcMsg rsp = {.handle = pRpcMsg->handle, .ahandle = pRpcMsg->ahandle, .code = code};
rpcSendResponse(&rsp);
}
rpcFreeCont(pRpcMsg->pCont);
}
mndCleanupMsg(pMsg);
}
static int32_t dndWriteMnodeMsgToQueue(SMnode *pMnode, taos_queue pQueue, SRpcMsg *pRpcMsg) {
SMnodeMsg *pMsg = mndInitMsg(pMnode, pRpcMsg);
if (pMsg == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
if (taosWriteQitem(pQueue, pMsg) != 0) {
mndCleanupMsg(pMsg);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
return 0;
}
void dndProcessMnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pWriteQ, pMsg) != 0) {
if (pMsg->msgType & 1u) {
SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno};
rpcSendResponse(&rsp);
}
rpcFreeCont(pMsg->pCont);
pMsg->pCont = NULL;
}
dndReleaseMnode(pDnode, pMnode);
dndWriteMnodeMsgToWorker(pDnode, &pDnode->mmgmt.writeWorker, pMsg);
}
void dndProcessMnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pSyncQ, pMsg) != 0) {
if (pMsg->msgType & 1u) {
SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno};
rpcSendResponse(&rsp);
}
rpcFreeCont(pMsg->pCont);
pMsg->pCont = NULL;
}
dndReleaseMnode(pDnode, pMnode);
dndWriteMnodeMsgToWorker(pDnode, &pDnode->mmgmt.syncWorker, pMsg);
}
void dndProcessMnodeReadMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pReadQ, pMsg) != 0) {
if (pMsg->msgType & 1u) {
SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno};
rpcSendResponse(&rsp);
}
rpcFreeCont(pMsg->pCont);
pMsg->pCont = NULL;
}
dndReleaseMnode(pDnode, pMnode);
}
static int32_t dndAllocMnodeReadQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->pReadQ = tWorkerAllocQueue(&pMgmt->readPool, pDnode, (FProcessItem)dndProcessMnodeReadQueue);
if (pMgmt->pReadQ == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
return 0;
}
static void dndFreeMnodeReadQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerFreeQueue(&pMgmt->readPool, pMgmt->pReadQ);
pMgmt->pReadQ = NULL;
}
static int32_t dndInitMnodeReadWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SWorkerPool *pPool = &pMgmt->readPool;
pPool->name = "mnode-read";
pPool->min = 0;
pPool->max = 1;
if (tWorkerInit(pPool) != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
dDebug("mnode read worker is initialized");
return 0;
}
static void dndCleanupMnodeReadWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerCleanup(&pMgmt->readPool);
dDebug("mnode read worker is closed");
}
static int32_t dndAllocMnodeWriteQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->pWriteQ = tWorkerAllocQueue(&pMgmt->writePool, pDnode, (FProcessItem)dndProcessMnodeWriteQueue);
if (pMgmt->pWriteQ == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
return 0;
}
static void dndFreeMnodeWriteQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerFreeQueue(&pMgmt->writePool, pMgmt->pWriteQ);
pMgmt->pWriteQ = NULL;
}
static int32_t dndInitMnodeWriteWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SWorkerPool *pPool = &pMgmt->writePool;
pPool->name = "mnode-write";
pPool->min = 0;
pPool->max = 1;
if (tWorkerInit(pPool) != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
dDebug("mnode write worker is initialized");
return 0;
}
static void dndCleanupMnodeWriteWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerCleanup(&pMgmt->writePool);
dDebug("mnode write worker is closed");
}
static int32_t dndAllocMnodeSyncQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->pSyncQ = tWorkerAllocQueue(&pMgmt->syncPool, pDnode, (FProcessItem)dndProcessMnodeSyncQueue);
if (pMgmt->pSyncQ == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
return 0;
}
static void dndFreeMnodeSyncQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerFreeQueue(&pMgmt->syncPool, pMgmt->pSyncQ);
pMgmt->pSyncQ = NULL;
}
static int32_t dndInitMnodeSyncWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SWorkerPool *pPool = &pMgmt->syncPool;
pPool->name = "mnode-sync";
pPool->min = 0;
pPool->max = 1;
if (tWorkerInit(pPool) != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
dDebug("mnode sync worker is initialized");
return 0;
}
static void dndCleanupMnodeSyncWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerCleanup(&pMgmt->syncPool);
dDebug("mnode sync worker is closed");
dndWriteMnodeMsgToWorker(pDnode, &pDnode->mmgmt.readWorker, pMsg);
}
int32_t dndInitMnode(SDnode *pDnode) {
@ -752,14 +534,6 @@ int32_t dndInitMnode(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
taosInitRWLatch(&pMgmt->latch);
char path[PATH_MAX];
snprintf(path, PATH_MAX, "%s/mnode.json", pDnode->dir.dnode);
pMgmt->file = strdup(path);
if (pMgmt->file == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
if (dndReadMnodeFile(pDnode) != 0) {
return -1;
}
@ -790,13 +564,13 @@ int32_t dndInitMnode(SDnode *pDnode) {
}
void dndCleanupMnode(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
dInfo("dnode-mnode start to clean up");
if (pMgmt->pMnode) dndStopMnodeWorker(pDnode);
tfree(pMgmt->file);
mndClose(pMgmt->pMnode);
pMgmt->pMnode = NULL;
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
if (pMgmt->pMnode) {
dndStopMnodeWorker(pDnode);
mndClose(pMgmt->pMnode);
pMgmt->pMnode = NULL;
}
dInfo("dnode-mnode is cleaned up");
}

18
source/dnode/mgmt/impl/src/dndQnode.c
View File

@ -140,26 +140,27 @@ static int32_t dndWriteQnodeFile(SDnode *pDnode) {
fclose(fp);
free(content);
if (taosRenameFile(file, file) != 0) {
char realfile[PATH_MAX + 20];
snprintf(realfile, PATH_MAX + 20, "%s/qnode.json", pDnode->dir.dnode);
if (taosRenameFile(file, realfile) != 0) {
terrno = TSDB_CODE_DND_QNODE_WRITE_FILE_ERROR;
dError("failed to rename %s since %s", file, terrstr());
return -1;
}
dInfo("successed to write %s, deployed:%d dropped:%d", file, pMgmt->deployed, pMgmt->dropped);
dInfo("successed to write %s, deployed:%d dropped:%d", realfile, pMgmt->deployed, pMgmt->dropped);
return 0;
}
static int32_t dndStartQnodeWorker(SDnode *pDnode) {
SQnodeMgmt *pMgmt = &pDnode->qmgmt;
if (dndInitWorker(pDnode, &pMgmt->queryWorker, DND_WORKER_SINGLE, "qnode-query", 0, 1,
(FProcessItem)dndProcessQnodeQueue) != 0) {
if (dndInitWorker(pDnode, &pMgmt->queryWorker, DND_WORKER_SINGLE, "qnode-query", 0, 1, dndProcessQnodeQueue) != 0) {
dError("failed to start qnode query worker since %s", terrstr());
return -1;
}
if (dndInitWorker(pDnode, &pMgmt->fetchWorker, DND_WORKER_SINGLE, "qnode-fetch", 0, 1,
(FProcessItem)dndProcessQnodeQueue) != 0) {
if (dndInitWorker(pDnode, &pMgmt->fetchWorker, DND_WORKER_SINGLE, "qnode-fetch", 0, 1, dndProcessQnodeQueue) != 0) {
dError("failed to start qnode fetch worker since %s", terrstr());
return -1;
}
@ -209,7 +210,9 @@ static int32_t dndOpenQnode(SDnode *pDnode) {
return -1;
}
pMgmt->deployed = 1;
if (dndWriteQnodeFile(pDnode) != 0) {
pMgmt->deployed = 0;
dError("failed to write qnode file since %s", terrstr());
dndStopQnodeWorker(pDnode);
qndClose(pQnode);
@ -218,7 +221,6 @@ static int32_t dndOpenQnode(SDnode *pDnode) {
taosWLockLatch(&pMgmt->latch);
pMgmt->pQnode = pQnode;
pMgmt->deployed = 1;
taosWUnLockLatch(&pMgmt->latch);
dInfo("qnode open successfully");
@ -250,6 +252,8 @@ static int32_t dndDropQnode(SDnode *pDnode) {
dndReleaseQnode(pDnode, pQnode);
dndStopQnodeWorker(pDnode);
pMgmt->deployed = 0;
dndWriteQnodeFile(pDnode);
qndClose(pQnode);
pMgmt->pQnode = NULL;

22
source/dnode/mgmt/impl/src/dndSnode.c
View File

@ -140,20 +140,22 @@ static int32_t dndWriteSnodeFile(SDnode *pDnode) {
fclose(fp);
free(content);
if (taosRenameFile(file, file) != 0) {
char realfile[PATH_MAX + 20];
snprintf(realfile, PATH_MAX + 20, "%s/snode.json", pDnode->dir.dnode);
if (taosRenameFile(file, realfile) != 0) {
terrno = TSDB_CODE_DND_SNODE_WRITE_FILE_ERROR;
dError("failed to rename %s since %s", file, terrstr());
return -1;
}
dInfo("successed to write %s, deployed:%d dropped:%d", file, pMgmt->deployed, pMgmt->dropped);
dInfo("successed to write %s, deployed:%d dropped:%d", realfile, pMgmt->deployed, pMgmt->dropped);
return 0;
}
static int32_t dndStartSnodeWorker(SDnode *pDnode) {
SSnodeMgmt *pMgmt = &pDnode->smgmt;
if (dndInitWorker(pDnode, &pMgmt->writeWorker, DND_WORKER_SINGLE, "snode-write", 0, 1,
(FProcessItem)dndProcessSnodeQueue) != 0) {
if (dndInitWorker(pDnode, &pMgmt->writeWorker, DND_WORKER_SINGLE, "snode-write", 0, 1, dndProcessSnodeQueue) != 0) {
dError("failed to start snode write worker since %s", terrstr());
return -1;
}
@ -202,7 +204,9 @@ static int32_t dndOpenSnode(SDnode *pDnode) {
return -1;
}
pMgmt->deployed = 1;
if (dndWriteSnodeFile(pDnode) != 0) {
pMgmt->deployed = 0;
dError("failed to write snode file since %s", terrstr());
dndStopSnodeWorker(pDnode);
sndClose(pSnode);
@ -211,7 +215,6 @@ static int32_t dndOpenSnode(SDnode *pDnode) {
taosWLockLatch(&pMgmt->latch);
pMgmt->pSnode = pSnode;
pMgmt->deployed = 1;
taosWUnLockLatch(&pMgmt->latch);
dInfo("snode open successfully");
@ -243,6 +246,8 @@ static int32_t dndDropSnode(SDnode *pDnode) {
dndReleaseSnode(pDnode, pSnode);
dndStopSnodeWorker(pDnode);
pMgmt->deployed = 0;
dndWriteSnodeFile(pDnode);
sndClose(pSnode);
pMgmt->pSnode = NULL;
sndDestroy(pDnode->dir.snode);
@ -328,7 +333,12 @@ int32_t dndInitSnode(SDnode *pDnode) {
return -1;
}
if (pMgmt->dropped) return 0;
if (pMgmt->dropped) {
dInfo("snode has been deployed and needs to be deleted");
sndDestroy(pDnode->dir.snode);
return 0;
}
if (!pMgmt->deployed) return 0;
return dndOpenSnode(pDnode);

19
source/dnode/mgmt/impl/src/dndWorker.c
View File

@ -73,7 +73,7 @@ void dndCleanupWorker(SDnodeWorker *pWorker) {
tWorkerCleanup(&pWorker->pool);
tWorkerFreeQueue(&pWorker->pool, pWorker->queue);
} else if (pWorker->type == DND_WORKER_MULTI) {
tWorkerCleanup(&pWorker->mpool);
tMWorkerCleanup(&pWorker->mpool);
tMWorkerFreeQueue(&pWorker->mpool, pWorker->queue);
} else {
}
@ -85,16 +85,23 @@ int32_t dndWriteMsgToWorker(SDnodeWorker *pWorker, void *pCont, int32_t contLen)
return -1;
}
void *pMsg = taosAllocateQitem(contLen);
void *pMsg = NULL;
if (contLen != 0) {
pMsg = taosAllocateQitem(contLen);
if (pMsg != NULL) {
memcpy(pMsg, pCont, contLen);
}
} else {
pMsg = pCont;
}
if (pMsg == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
memcpy(pMsg, pCont, contLen);
if (taosWriteQitem(pWorker, pMsg) != 0) {
taosFreeItem(pMsg);
if (taosWriteQitem(pWorker->queue, pMsg) != 0) {
taosFreeQitem(pMsg);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}

2
source/dnode/mgmt/impl/test/dnode/dnode.cpp
View File

@ -162,7 +162,7 @@ TEST_F(DndTestDnode, 03_Create_Drop_Restart_Dnode) {
SCreateDnodeMsg* pReq = (SCreateDnodeMsg*)rpcMallocCont(contLen);
strcpy(pReq->fqdn, "localhost");
pReq->port = htonl(904);
pReq->port = htonl(9044);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_DNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);

2
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -388,7 +388,7 @@ static int32_t mndCreateDnode(SMnode *pMnode, SMnodeMsg *pMsg, SCreateDnodeMsg *
dnodeObj.updateTime = dnodeObj.createdTime;
dnodeObj.port = pCreate->port;
memcpy(dnodeObj.fqdn, pCreate->fqdn, TSDB_FQDN_LEN);
snprintf(dnodeObj.ep, "%s:%u", dnodeObj.fqdn, dnodeObj.port);
snprintf(dnodeObj.ep, TSDB_EP_LEN, "%s:%u", dnodeObj.fqdn, dnodeObj.port);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, &pMsg->rpcMsg);
if (pTrans == NULL) {

8
source/dnode/mnode/impl/src/mnode.c
View File

@ -390,7 +390,7 @@ void mndSendRsp(SMnodeMsg *pMsg, int32_t code) {
rpcSendResponse(&rpcRsp);
}
static void mndProcessRpcMsg(SMnodeMsg *pMsg) {
void mndProcessMsg(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
int32_t code = 0;
tmsg_t msgType = pMsg->rpcMsg.msgType;
@ -451,12 +451,6 @@ void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp) {
}
}
void mndProcessReadMsg(SMnodeMsg *pMsg) { mndProcessRpcMsg(pMsg); }
void mndProcessWriteMsg(SMnodeMsg *pMsg) { mndProcessRpcMsg(pMsg); }
void mndProcessSyncMsg(SMnodeMsg *pMsg) { mndProcessRpcMsg(pMsg); }
uint64_t mndGenerateUid(char *name, int32_t len) {
int64_t us = taosGetTimestampUs();
int32_t hashval = MurmurHash3_32(name, len);

1
source/libs/parser/src/astToMsg.c
View File

@ -428,6 +428,7 @@ SDropDnodeMsg *buildDropDnodeMsg(SSqlInfo* pInfo, int32_t* len, SMsgBuf* pMsgBuf
char* end = NULL;
SDropDnodeMsg * pDrop = (SDropDnodeMsg *)calloc(1, sizeof(SDropDnodeMsg));
pDrop->dnodeId = strtoll(pzName->z, &end, 10);
pDrop->dnodeId = htonl(pDrop->dnodeId);
*len = sizeof(SDropDnodeMsg);
if (end - pzName->z != pzName->n) {

11
tests/script/unique/dnode/basic1.sim
View File

@ -94,5 +94,16 @@ if $rows != 2 then
return -1
endi
print =============== drop dnode
sql drop dnode 2;
sql show dnodes;
if $rows != 1 then
return -1
endi
if $data00 != 1 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode2 -s stop -x SIGINT