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

change super table vgroup

This commit is contained in:
slguan 2020-04-16 13:31:56 +08:00
parent d7d51ad10b
commit de76b5bd30
3 changed files with 35 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/inc/mnode.h
View File

@ -88,6 +88,7 @@ typedef struct _dnode_obj {
typedef struct {
int32_t dnodeId;
uint16_t port;
uint32_t privateIp;
uint32_t publicIp;
} SVnodeGid;

11
src/inc/taosmsg.h
View File

@ -634,14 +634,13 @@ typedef struct SCMSTableVgroupMsg {
} SCMSTableVgroupMsg;
typedef struct {
SIpAddr ipAddr;
int32_t numOfVgroups;
int32_t vgId[];
} STableDnodeVgroupInfo;
int32_t vgId;
SIpAddr ipAddr[TSDB_REPLICA_MAX_NUM];
} SCMVgroupInfo;
typedef struct {
int32_t numOfDnodes;
STableDnodeVgroupInfo dnodeVgroups[];
int32_t numOfVgroups;
SCMVgroupInfo vgroups[];
} SCMSTableVgroupRspMsg;
typedef struct {

50
src/mnode/src/mgmtTable.c
View File

@ -1166,40 +1166,48 @@ static void mgmtGetSuperTableMeta(SQueuedMsg *pMsg) {
static void mgmtProcessSuperTableVgroupMsg(SQueuedMsg *pMsg) {
SCMSTableVgroupMsg *pInfo = pMsg->pCont;
pMsg->pTable = mgmtGetSuperTable(pInfo->tableId);
SSuperTableObj *pTable = mgmtGetSuperTable(pInfo->tableId);
pMsg->pTable = pTable;
if (pMsg->pTable == NULL) {
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_TABLE);
return;
}
SCMSTableVgroupRspMsg *pRsp = rpcMallocCont(sizeof(SCMSTableVgroupRspMsg) + sizeof(uint32_t) * clusterGetDnodesNum());
int32_t contLen = sizeof(SCMSTableVgroupRspMsg) + sizeof(SCMVgroupInfo) * pTable->vgLen;
SCMSTableVgroupRspMsg *pRsp = rpcMallocCont(contLen);
if (pRsp == NULL) {
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_TABLE);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SERV_OUT_OF_MEMORY);
return;
}
int32_t numOfVgroups = 1;
int32_t numOfDnodes = 1;
pRsp->numOfDnodes = htonl(numOfDnodes);
STableDnodeVgroupInfo* pVgroupInfo = pRsp->dnodeVgroups;
pVgroupInfo->ipAddr.ip = htonl(inet_addr(tsPrivateIp));
pVgroupInfo->ipAddr.port = htons(0); // todo fix it
pVgroupInfo->numOfVgroups = htonl(numOfVgroups); // todo fix it
int32_t* vgIdList = pVgroupInfo->vgId;
for(int32_t i = 0; i < numOfVgroups; ++i) {
vgIdList[i] = htonl(2); // todo fix it
int32_t vg = 0;
for (; vg < pTable->vgLen; ++vg) {
int32_t vgId = pTable->vgList[vg];
if (vgId == 0) break;
SVgObj *pVgroup = mgmtGetVgroup(vgId);
if (pVgroup == NULL) break;
pRsp->vgroups[vg].vgId = htonl(vgId);
for (int32_t vn = 0; vn < pVgroup->numOfVnodes; ++vn) {
SDnodeObj *pDnode = clusterGetDnode(pVgroup->vnodeGid[vn].dnodeId);
if (pDnode == NULL) break;
pRsp->vgroups[vg].ipAddr[vn].ip = htonl(pDnode->privateIp);
pRsp->vgroups[vg].ipAddr[vn].port = htons(tsDnodeShellPort);
clusterReleaseDnode(pDnode);
}
mgmtReleaseVgroup(pVgroup);
}
assert(numOfDnodes == 1); // this size is valid only when numOfDnodes equals 1
int32_t msgLen = sizeof(SCMSTableVgroupRspMsg) + sizeof(STableDnodeVgroupInfo) + numOfVgroups * sizeof(int32_t);
pRsp->numOfVgroups = htonl(vg);
SRpcMsg rpcRsp = {0};
rpcRsp.handle = pMsg->thandle;
rpcRsp.pCont = pRsp;
rpcRsp.contLen = msgLen;
rpcRsp.contLen = sizeof(SCMSTableVgroupRspMsg) + sizeof(SCMVgroupInfo) * vg;
rpcSendResponse(&rpcRsp);
}