Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

enh: dmodule API dependence and grant process #23422

Merged
merged 10 commits into from
Oct 30, 2023
Merged

enh: dmodule API dependence and grant process #23422

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
7bda08c
Merge branch '3.0' into enh/TD-25879-3.0x
kailixu Sep 27, 2023
9ddb246
fix: remove asserts for x86/arm
kailixu Oct 7, 2023
14a9b77
chore: merge 3.0
kailixu Oct 26, 2023
96de7b0
chore: update grant/dnode mgmt
kailixu Oct 26, 2023
9f63d58
chore: update dmoudle for darwin m1/x86/win x64
kailixu Oct 26, 2023
ef6bdec
chore: update dmodule for arm/darwin x64
kailixu Oct 27, 2023
9ae8345
chore: code format
kailixu Oct 27, 2023
3f65773
enh: send grant to dnodes online and in async mode
kailixu Oct 28, 2023
22d6b95
enh: dm init API
kailixu Oct 29, 2023
b0d0cd5
chore: update dmodule lib
kailixu Oct 29, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
Binary file modified deps/arm/dm_static/libdmodule.a
View file
Open in desktop
Binary file not shown.
Binary file modified deps/darwin/arm/dm_static/libdmodule.a
View file
Open in desktop
Binary file not shown.
Binary file modified deps/darwin/x64/dm_static/libdmodule.a
View file
Open in desktop
Binary file not shown.
Binary file modified deps/win/x64/dm_static/dmodule.lib
View file
Open in desktop
Binary file not shown.
Binary file modified deps/x86/dm_static/libdmodule.a
View file
Open in desktop
Binary file not shown.
4 changes: 0 additions & 4 deletions include/common/tgrant.h
View file
Open in desktop
Expand Up @@ -51,11 +51,7 @@ typedef enum {
} EGrantType;

int32_t grantCheck(EGrantType grant);
#ifndef TD_GRANT_OPTIMIZE
int32_t grantAlterActiveCode(const char* old, const char* newer, char* out, int8_t type);
#else
int32_t grantAlterActiveCode(int32_t did, const char* old, const char* newer, char* out, int8_t type);
#endif

#ifndef GRANTS_CFG
#ifdef TD_ENTERPRISE
Expand Down
3 changes: 3 additions & 0 deletions include/common/tmsg.h
View file
Open in desktop
Expand Up @@ -1568,6 +1568,9 @@ typedef struct {
typedef struct {
int32_t id;
int8_t isMnode;
#ifdef TD_GRANT_HB_OPTIMIZE
int8_t offlineReason;
#endif
SEp ep;
char active[TSDB_ACTIVE_KEY_LEN];
char connActive[TSDB_CONN_ACTIVE_KEY_LEN];
Expand Down
2 changes: 1 addition & 1 deletion source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
View file
Open in desktop
Expand Up @@ -163,7 +163,7 @@ SArray *mmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_STREAM, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_PAUSE_STREAM, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_RESUME_STREAM, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT_RSP, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;

if (dmSetMgmtHandle(pArray, TDMT_MND_RETRIEVE_IP_WHITE, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_GET_USER_WHITELIST, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
Expand Down
8 changes: 8 additions & 0 deletions source/dnode/mgmt/node_mgmt/inc/dmMgmt.h
View file
Open in desktop
Expand Up @@ -97,7 +97,11 @@ int32_t dmMarkWrapper(SMgmtWrapper *pWrapper);
void dmReleaseWrapper(SMgmtWrapper *pWrapper);
int32_t dmInitVars(SDnode *pDnode);
void dmClearVars(SDnode *pDnode);
#if defined(TD_MODULE_OPTIMIZE) || !defined(TD_ENTERPRISE)
int32_t dmInitModule(SDnode *pDnode, SMgmtWrapper *wrappers);
#else
int32_t dmInitModule(SDnode *pDnode);
#endif
bool dmRequireNode(SDnode *pDnode, SMgmtWrapper *pWrapper);
SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper);
void dmSetStatus(SDnode *pDnode, EDndRunStatus stype);
Expand All @@ -119,7 +123,11 @@ int32_t dmInitStatusClient(SDnode *pDnode);
void dmCleanupClient(SDnode *pDnode);
void dmCleanupStatusClient(SDnode *pDnode);
SMsgCb dmGetMsgcb(SDnode *pDnode);
#if defined(TD_MODULE_OPTIMIZE) || !defined(TD_ENTERPRISE)
int32_t dmInitMsgHandle(SDnode *pDnode, SMgmtWrapper *wrappers);
#else
int32_t dmInitMsgHandle(SDnode *pDnode);
#endif
int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);

// dmMonitor.c
Expand Down
77 changes: 77 additions & 0 deletions source/dnode/mgmt/node_mgmt/src/dmMgmt.c
View file
Open in desktop
Expand Up @@ -66,9 +66,15 @@ int32_t dmInitDnode(SDnode *pDnode) {
goto _OVER;
}

#if defined(TD_MODULE_OPTIMIZE) || !defined(TD_ENTERPRISE)
if (dmInitModule(pDnode, pDnode->wrappers) != 0) {
goto _OVER;
}
#else
if (dmInitModule(pDnode) != 0) {
goto _OVER;
}
#endif

indexInit(tsNumOfCommitThreads);
streamMetaInit();
Expand Down Expand Up @@ -107,6 +113,77 @@ void dmCleanupDnode(SDnode *pDnode) {
dDebug("dnode is closed, ptr:%p", pDnode);
}

#if defined(TD_MODULE_OPTIMIZE) || !defined(TD_ENTERPRISE)
int32_t dmInitVars(SDnode *pDnode) {
SDnodeData *pData = &pDnode->data;
pData->dnodeId = 0;
pData->clusterId = 0;
pData->dnodeVer = 0;
pData->engineVer = 0;
pData->updateTime = 0;
pData->rebootTime = taosGetTimestampMs();
pData->dropped = 0;
pData->stopped = 0;

pData->dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
if (pData->dnodeHash == NULL) {
dError("failed to init dnode hash");
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}

if (dmReadEps(pData) != 0) {
dError("failed to read file since %s", terrstr());
return -1;
}

if (pData->dropped) {
dError("dnode will not start since its already dropped");
return -1;
}

taosThreadRwlockInit(&pData->lock, NULL);
taosThreadMutexInit(&pDnode->mutex, NULL);
return 0;
}

void dmClearVars(SDnode *pDnode) {
for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
taosMemoryFreeClear(pWrapper->path);
taosThreadRwlockDestroy(&pWrapper->lock);
}
if (pDnode->lockfile != NULL) {
taosUnLockFile(pDnode->lockfile);
taosCloseFile(&pDnode->lockfile);
pDnode->lockfile = NULL;
}

SDnodeData *pData = &pDnode->data;
taosThreadRwlockWrlock(&pData->lock);
if (pData->oldDnodeEps != NULL) {
if (dmWriteEps(pData) == 0) {
dmRemoveDnodePairs(pData);
}
taosArrayDestroy(pData->oldDnodeEps);
pData->oldDnodeEps = NULL;
}
if (pData->dnodeEps != NULL) {
taosArrayDestroy(pData->dnodeEps);
pData->dnodeEps = NULL;
}
if (pData->dnodeHash != NULL) {
taosHashCleanup(pData->dnodeHash);
pData->dnodeHash = NULL;
}
taosThreadRwlockUnlock(&pData->lock);

taosThreadRwlockDestroy(&pData->lock);
taosThreadMutexDestroy(&pDnode->mutex);
memset(&pDnode->mutex, 0, sizeof(pDnode->mutex));
}
#endif

void dmSetStatus(SDnode *pDnode, EDndRunStatus status) {
if (pDnode->status != status) {
dDebug("dnode status set from %s to %s", dmStatStr(pDnode->status), dmStatStr(status));
Expand Down
28 changes: 28 additions & 0 deletions source/dnode/mgmt/node_mgmt/src/dmTransport.c
View file
Open in desktop
Expand Up @@ -251,6 +251,33 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
dmReleaseWrapper(pWrapper);
}

#if defined(TD_MODULE_OPTIMIZE) || !defined(TD_ENTERPRISE)
int32_t dmInitMsgHandle(SDnode *pDnode, SMgmtWrapper *wrappers) {
SDnodeTrans *pTrans = &pDnode->trans;

for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
SMgmtWrapper *pWrapper = wrappers + ntype;
SArray *pArray = (*pWrapper->func.getHandlesFp)();
if (pArray == NULL) return -1;

for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
SMgmtHandle *pMgmt = taosArrayGet(pArray, i);
SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pMgmt->msgType)];
if (pMgmt->needCheckVgId) {
pHandle->needCheckVgId = pMgmt->needCheckVgId;
}
if (!pMgmt->needCheckVgId) {
pHandle->defaultNtype = ntype;
}
pWrapper->msgFps[TMSG_INDEX(pMgmt->msgType)] = pMgmt->msgFp;
}

taosArrayDestroy(pArray);
}

return 0;
}
#else
int32_t dmInitMsgHandle(SDnode *pDnode) {
SDnodeTrans *pTrans = &pDnode->trans;

Expand All @@ -276,6 +303,7 @@ int32_t dmInitMsgHandle(SDnode *pDnode) {

return 0;
}
#endif

static inline int32_t dmSendReq(const SEpSet *pEpSet, SRpcMsg *pMsg) {
SDnode *pDnode = dmInstance();
Expand Down
11 changes: 3 additions & 8 deletions source/dnode/mnode/impl/src/mndDnode.c
View file
Open in desktop
Expand Up @@ -397,6 +397,9 @@ void mndGetDnodeData(SMnode *pMnode, SArray *pDnodeInfo) {
SDnodeInfo dInfo;
dInfo.id = pDnode->id;
dInfo.ep.port = pDnode->port;
#ifdef TD_GRANT_HB_OPTIMIZE
dInfo.offlineReason = pDnode->offlineReason;
#endif
tstrncpy(dInfo.ep.fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
tstrncpy(dInfo.active, pDnode->active, TSDB_ACTIVE_KEY_LEN);
tstrncpy(dInfo.connActive, pDnode->connActive, TSDB_CONN_ACTIVE_KEY_LEN);
Expand Down Expand Up @@ -781,11 +784,7 @@ static int32_t mndConfigDnode(SMnode *pMnode, SRpcMsg *pReq, SMCfgDnodeReq *pCfg

SDnodeObj tmpDnode = *pDnode;
if (action == DND_ACTIVE_CODE) {
#ifndef TD_GRANT_OPTIMIZE
if (grantAlterActiveCode(pDnode->active, pCfgReq->value, tmpDnode.active, 0) != 0) {
#else
if (grantAlterActiveCode(pDnode->id, pDnode->active, pCfgReq->value, tmpDnode.active, 0) != 0) {
#endif
if (TSDB_CODE_DUP_KEY != terrno) {
mError("dnode:%d, config dnode:%d, app:%p config:%s value:%s failed since %s", pDnode->id, pCfgReq->dnodeId,
pReq->info.ahandle, pCfgReq->config, pCfgReq->value, terrstr());
Expand All @@ -801,11 +800,7 @@ static int32_t mndConfigDnode(SMnode *pMnode, SRpcMsg *pReq, SMCfgDnodeReq *pCfg
goto _OVER;
}
} else if (action == DND_CONN_ACTIVE_CODE) {
#ifndef TD_GRANT_OPTIMIZE
if (grantAlterActiveCode(pDnode->connActive, pCfgReq->value, tmpDnode.connActive, 1) != 0) {
#else
if (grantAlterActiveCode(pDnode->id, pDnode->connActive, pCfgReq->value, tmpDnode.connActive, 1) != 0) {
#endif
if (TSDB_CODE_DUP_KEY != terrno) {
mError("dnode:%d, config dnode:%d, app:%p config:%s value:%s failed since %s", pDnode->id, pCfgReq->dnodeId,
pReq->info.ahandle, pCfgReq->config, pCfgReq->value, terrstr());
Expand Down
4 changes: 0 additions & 4 deletions source/dnode/mnode/impl/src/mndGrant.c
View file
Open in desktop
Expand Up @@ -131,13 +131,9 @@ void grantAdd(EGrantType grant, uint64_t value) {}
void grantRestore(EGrantType grant, uint64_t value) {}
int32_t dmProcessGrantReq(void *pInfo, SRpcMsg *pMsg) { return TSDB_CODE_SUCCESS; }
int32_t dmProcessGrantNotify(void *pInfo, SRpcMsg *pMsg) { return TSDB_CODE_SUCCESS; }
#ifndef TD_GRANT_OPTIMIZE
int32_t grantAlterActiveCode(const char *old, const char *new, char *out, int8_t type) { return TSDB_CODE_SUCCESS; }
#else
int32_t grantAlterActiveCode(int32_t did, const char *old, const char *new, char *out, int8_t type) {
return TSDB_CODE_SUCCESS;
}
#endif

#endif

Expand Down