提交 90e94473 编写于 作者: S Shengliang Guan

adjust vnode queue

上级 1e1ef0e0
......@@ -24,6 +24,8 @@
extern "C" {
#endif
typedef enum { VND_WRITE_QUEUE, VND_QUERY_QUEUE, VND_FETCH_QUEUE, VND_APPLY_QUEUE, VND_SYNC_QUEUE } EVndQueueType;
typedef struct SVnodesMgmt {
SHashObj *hash;
SRWLatch latch;
......@@ -102,7 +104,8 @@ int32_t vmAllocQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode);
void vmFreeQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode);
int32_t vmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, int32_t vgId, SRpcMsg *pMsg);
int32_t vmPutMsgToFetchQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
int32_t vmProcessWriteMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg);
int32_t vmProcessSyncMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg);
......
......@@ -296,7 +296,10 @@ static int32_t vmInit(SMgmtWrapper *pWrapper) {
vnodeOpt.nthreads = tsNumOfCommitThreads;
vnodeOpt.putToQueryQFp = vmPutMsgToQueryQueue;
vnodeOpt.putToFetchQFp = vmPutMsgToQueryQueue;
vnodeOpt.sendReqFp = dndSendReqToDnode;
vnodeOpt.sendMnodeReqFp = dndSendReqToMnode;
vnodeOpt.sendRspFp = dndSendRsp;
if (vnodeInit(&vnodeOpt) != 0) {
dError("failed to init vnode since %s", terrstr());
goto _OVER;
......
......@@ -16,6 +16,43 @@
#define _DEFAULT_SOURCE
#include "vmInt.h"
static void vmProcessMgmtQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
int32_t code = -1;
tmsg_t msgType = pMsg->rpcMsg.msgType;
dTrace("msg:%p, will be processed in vnode-mgmt queue", pMsg);
switch (msgType) {
case TDMT_DND_CREATE_VNODE:
code = vmProcessCreateVnodeReq(pMgmt, pMsg);
break;
case TDMT_DND_ALTER_VNODE:
code = vmProcessAlterVnodeReq(pMgmt, pMsg);
break;
case TDMT_DND_DROP_VNODE:
code = vmProcessDropVnodeReq(pMgmt, pMsg);
break;
case TDMT_DND_SYNC_VNODE:
code = vmProcessSyncVnodeReq(pMgmt, pMsg);
break;
case TDMT_DND_COMPACT_VNODE:
code = vmProcessCompactVnodeReq(pMgmt, pMsg);
break;
default:
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
dError("msg:%p, not processed in vnode-mgmt queue", pMsg);
}
if (msgType & 1u) {
if (code != 0 && terrno != 0) code = terrno;
SRpcMsg rsp = {.code = code, .handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle};
dndSendRsp(pMgmt->pWrapper, &rsp);
}
dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
rpcFreeCont(pMsg->rpcMsg.pCont);
taosFreeQitem(pMsg);
}
static void vmProcessQueryQueue(SVnodeObj *pVnode, SNodeMsg *pMsg) {
dTrace("msg:%p, will be processed in vnode query queue", pMsg);
vnodeProcessQueryMsg(pVnode->pImpl, &pMsg->rpcMsg);
......@@ -89,93 +126,112 @@ static void vmProcessSyncQueue(SVnodeObj *pVnode, STaosQall *qall, int32_t numOf
}
}
static SVnodeObj *vmAcquireFromMsg(SVnodesMgmt *pMgmt, SNodeMsg *pNodeMsg) {
SRpcMsg *pMsg = &pNodeMsg->rpcMsg;
static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EVndQueueType qtype) {
SRpcMsg *pRpc = &pMsg->rpcMsg;
int32_t code = -1;
SMsgHead *pHead = pMsg->pCont;
SMsgHead *pHead = pRpc->pCont;
pHead->contLen = htonl(pHead->contLen);
pHead->vgId = htonl(pHead->vgId);
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
if (pVnode == NULL) {
dError("vgId:%d, failed to acquire vnode while process req", pHead->vgId);
dError("vgId:%d, failed to write msg:%p to queue since %s", pHead->vgId, pMsg, terrstr());
return -1;
}
return pVnode;
}
int32_t vmProcessWriteMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
SVnodeObj *pVnode = vmAcquireFromMsg(pMgmt, pMsg);
if (pVnode == NULL) return -1;
switch (qtype) {
case VND_QUERY_QUEUE:
dTrace("msg:%p, will be written into vnode-query queue", pMsg);
code = taosWriteQitem(pVnode->pQueryQ, pMsg);
break;
case VND_FETCH_QUEUE:
dTrace("msg:%p, will be written into vnode-fetch queue", pMsg);
code = taosWriteQitem(pVnode->pFetchQ, pMsg);
break;
case VND_WRITE_QUEUE:
dTrace("msg:%p, will be written into vnode-write queue", pMsg);
code = taosWriteQitem(pVnode->pWriteQ, pMsg);
case VND_SYNC_QUEUE:
dTrace("msg:%p, will be written into vnode-sync queue", pMsg);
code = taosWriteQitem(pVnode->pSyncQ, pMsg);
default:
terrno = TSDB_CODE_INVALID_PARA;
break;
}
int32_t code = taosWriteQitem(pVnode->pWriteQ, pMsg);
vmReleaseVnode(pMgmt, pVnode);
return code;
}
int32_t vmProcessSyncMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
SVnodeObj *pVnode = vmAcquireFromMsg(pMgmt, pMsg);
if (pVnode == NULL) return -1;
return vmPutNodeMsgToQueue(pMgmt, pMsg, VND_SYNC_QUEUE);
}
int32_t code = taosWriteQitem(pVnode->pSyncQ, pMsg);
vmReleaseVnode(pMgmt, pVnode);
return code;
int32_t vmProcessWriteMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
return vmPutNodeMsgToQueue(pMgmt, pMsg, VND_WRITE_QUEUE);
}
int32_t vmProcessQueryMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
SVnodeObj *pVnode = vmAcquireFromMsg(pMgmt, pMsg);
if (pVnode == NULL) return -1;
int32_t code = taosWriteQitem(pVnode->pQueryQ, pMsg);
vmReleaseVnode(pMgmt, pVnode);
return code;
return vmPutNodeMsgToQueue(pMgmt, pMsg, VND_QUERY_QUEUE);
}
int32_t vmProcessFetchMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
SVnodeObj *pVnode = vmAcquireFromMsg(pMgmt, pMsg);
if (pVnode == NULL) return -1;
return vmPutNodeMsgToQueue(pMgmt, pMsg, VND_FETCH_QUEUE);
}
int32_t code = taosWriteQitem(pVnode->pFetchQ, pMsg);
vmReleaseVnode(pMgmt, pVnode);
return code;
int32_t vmProcessMgmtMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
SDnodeWorker *pWorker = &pMgmt->mgmtWorker;
dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name);
return dndWriteMsgToWorker(pWorker, pMsg);
}
int32_t vmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EVndQueueType qtype) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
int32_t code = -1;
SMsgHead *pHead = pRpc->pCont;
// pHead->vgId = htonl(pHead->vgId);
int32_t code = -1;
SMsgHead *pHead = pRpc->pCont;
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
if (pVnode == NULL) return -1;
SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
if (pMsg != NULL) {
dTrace("msg:%p, is created, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
pMsg->rpcMsg = *pRpc;
code = taosWriteQitem(pVnode->pQueryQ, pMsg);
switch (qtype) {
case VND_QUERY_QUEUE:
dTrace("msg:%p, will be put into vnode-query queue", pMsg);
code = taosWriteQitem(pVnode->pQueryQ, pMsg);
break;
case VND_FETCH_QUEUE:
dTrace("msg:%p, will be put into vnode-fetch queue", pMsg);
code = taosWriteQitem(pVnode->pFetchQ, pMsg);
break;
case VND_APPLY_QUEUE:
dTrace("msg:%p, will be put into vnode-apply queue", pMsg);
code = taosWriteQitem(pVnode->pApplyQ, pMsg);
break;
case VND_WRITE_QUEUE:
case VND_SYNC_QUEUE:
default:
terrno = TSDB_CODE_INVALID_PARA;
break;
}
}
vmReleaseVnode(pMgmt, pVnode);
return code;
}
int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, int32_t vgId, SRpcMsg *pRpc) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
int32_t code = -1;
SMsgHead *pHead = pRpc->pCont;
// pHead->vgId = htonl(pHead->vgId);
int32_t vmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
return vmPutRpcMsgToQueue(pWrapper, pRpc, VND_QUERY_QUEUE);
}
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
if (pVnode == NULL) return -1;
int32_t vmPutMsgToFetchQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
return vmPutRpcMsgToQueue(pWrapper, pRpc, VND_FETCH_QUEUE);
}
SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
if (pMsg != NULL) {
pMsg->rpcMsg = *pRpc;
code = taosWriteQitem(pVnode->pApplyQ, pMsg);
}
vmReleaseVnode(pMgmt, pVnode);
return code;
int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
return vmPutRpcMsgToQueue(pWrapper, pRpc, VND_APPLY_QUEUE);
}
int32_t vmAllocQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode) {
......@@ -191,6 +247,7 @@ int32_t vmAllocQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode) {
return -1;
}
dDebug("vgId:%d, vnode queue is alloced", pVnode->vgId);
return 0;
}
......@@ -205,43 +262,7 @@ void vmFreeQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode) {
pVnode->pSyncQ = NULL;
pVnode->pFetchQ = NULL;
pVnode->pQueryQ = NULL;
}
static void vmProcessMgmtQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
int32_t code = -1;
tmsg_t msgType = pMsg->rpcMsg.msgType;
dTrace("msg:%p, will be processed in vnode-mgmt queue", pMsg);
switch (msgType) {
case TDMT_DND_CREATE_VNODE:
code = vmProcessCreateVnodeReq(pMgmt, pMsg);
break;
case TDMT_DND_ALTER_VNODE:
code = vmProcessAlterVnodeReq(pMgmt, pMsg);
break;
case TDMT_DND_DROP_VNODE:
code = vmProcessDropVnodeReq(pMgmt, pMsg);
break;
case TDMT_DND_SYNC_VNODE:
code = vmProcessSyncVnodeReq(pMgmt, pMsg);
break;
case TDMT_DND_COMPACT_VNODE:
code = vmProcessCompactVnodeReq(pMgmt, pMsg);
break;
default:
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
dError("msg:%p, not processed in mgmt queue", pMsg);
}
if (msgType & 1u) {
if (code != 0) code = terrno;
SRpcMsg rsp = {.code = code, .handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle};
dndSendRsp(pMgmt->pWrapper, &rsp);
}
dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
rpcFreeCont(pMsg->rpcMsg.pCont);
taosFreeQitem(pMsg);
dDebug("vgId:%d, vnode queue is freed", pVnode->vgId);
}
int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
......@@ -275,7 +296,7 @@ int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
if (tWWorkerInit(pWPool) != 0) return -1;
if (dndInitWorker(pMgmt, &pMgmt->mgmtWorker, DND_WORKER_SINGLE, "vnode-mgmt", 1, 1, vmProcessMgmtQueue) != 0) {
dError("failed to start dnode mgmt worker since %s", terrstr());
dError("failed to start vnode-mgmt worker since %s", terrstr());
return -1;
}
......@@ -291,9 +312,3 @@ void vmStopWorker(SVnodesMgmt *pMgmt) {
tWWorkerCleanup(&pMgmt->syncPool);
dDebug("vnode workers is closed");
}
int32_t vmProcessMgmtMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
SDnodeWorker *pWorker = &pMgmt->mgmtWorker;
dTrace("msg:%p, will be written to worker %s", pMsg, pWorker->name);
return dndWriteMsgToWorker(pWorker, pMsg);
}
\ No newline at end of file
......@@ -62,6 +62,7 @@ typedef struct {
typedef struct {
uint16_t nthreads; // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
PutToQueueFp putToQueryQFp;
PutToQueueFp putToFetchQFp;
SendReqFp sendReqFp;
SendMnodeReqFp sendMnodeReqFp;
SendRspFp sendRspFp;
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册