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

shm

This commit is contained in:
Shengliang Guan 2022-03-31 15:11:09 +08:00
parent b5ec7d28d9
commit 1a8b1753fb
11 changed files with 49 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/mgmt/dm/src/dmInt.c
View File

@ -74,7 +74,7 @@ void dmSendRedirectRsp(SDnodeMgmt *pMgmt, const SRpcMsg *pReq) {
}
static int32_t dmStart(SMgmtWrapper *pWrapper) {
dDebug("dnode-mgmt start to run");
dDebug("dnode-mgmt starts running");
return dmStartThread(pWrapper->pMgmt);
}

2
source/dnode/mgmt/dm/src/dmMsg.c
View File

@ -53,7 +53,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
tSerializeSStatusReq(pHead, contLen, &req);
taosArrayDestroy(req.pVloads);
SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)9527};
SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)0x9527};
pMgmt->statusSent = 1;
dTrace("send req:%s to mnode, app:%p", TMSG_INFO(rpcMsg.msgType), rpcMsg.ahandle);

22
source/dnode/mgmt/main/exe/dndMain.c
View File

@ -30,7 +30,7 @@ static struct {
} global = {0};
static void dndStopDnode(int signum, void *info, void *ctx) {
dInfo("signal:%d is received", signum);
dInfo("system signal:%d received", signum);
SDnode *pDnode = atomic_val_compare_exchange_ptr(&global.pDnode, 0, global.pDnode);
if (pDnode != NULL) {
dndHandleEvent(pDnode, DND_EVENT_STOP);
@ -38,8 +38,10 @@ static void dndStopDnode(int signum, void *info, void *ctx) {
}
static void dndHandleChild(int signum, void *info, void *ctx) {
dInfo("signal:%d is received", signum);
dndHandleEvent(global.pDnode, DND_EVENT_CHILD);
dInfo("sigchild received");
if (global.pDnode != NULL) {
dndHandleEvent(global.pDnode, DND_EVENT_CHILD);
}
}
static void dndSetSignalHandle() {
@ -50,7 +52,7 @@ static void dndSetSignalHandle() {
taosSetSignal(SIGBREAK, dndStopDnode);
if (!tsMultiProcess) {
} else if (global.ntype == DNODE) {
} else if (global.ntype == DNODE || global.ntype == NODE_MAX) {
taosSetSignal(SIGCHLD, dndHandleChild);
} else {
taosKillChildOnParentStopped();
@ -74,14 +76,14 @@ static int32_t dndParseArgs(int32_t argc, char const *argv[]) {
tstrncpy(global.apolloUrl, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-e") == 0) {
tstrncpy(global.envFile, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-k") == 0) {
global.generateGrant = true;
} else if (strcmp(argv[i], "-n") == 0) {
global.ntype = atoi(argv[++i]);
if (global.ntype <= DNODE || global.ntype > NODE_MAX) {
printf("'-n' range is [1-5], default is 0\n");
return -1;
}
} else if (strcmp(argv[i], "-k") == 0) {
global.generateGrant = true;
} else if (strcmp(argv[i], "-C") == 0) {
global.dumpConfig = true;
} else if (strcmp(argv[i], "-V") == 0) {
@ -139,7 +141,7 @@ static int32_t dndInitLog() {
static void dndSetProcInfo(int32_t argc, char **argv) {
taosSetProcPath(argc, argv);
if (global.ntype != DNODE) {
if (global.ntype != DNODE && global.ntype != NODE_MAX) {
const char *name = dndNodeProcStr(global.ntype);
taosSetProcName(argc, argv, name);
}
@ -147,14 +149,14 @@ static void dndSetProcInfo(int32_t argc, char **argv) {
static int32_t dndRunDnode() {
if (dndInit() != 0) {
dError("failed to initialize environment since %s", terrstr());
dError("failed to init environment since %s", terrstr());
return -1;
}
SDnodeOpt option = dndGetOpt();
SDnode *pDnode = dndCreate(&option);
if (pDnode == NULL) {
dError("failed to to create dnode object since %s", terrstr());
dError("failed to to create dnode since %s", terrstr());
return -1;
} else {
global.pDnode = pDnode;
@ -184,7 +186,6 @@ int main(int argc, char const *argv[]) {
return -1;
}
dndSetProcInfo(argc, (char **)argv);
if (global.generateGrant) {
dndGenerateGrant();
return 0;
@ -212,5 +213,6 @@ int main(int argc, char const *argv[]) {
return 0;
}
dndSetProcInfo(argc, (char **)argv);
return dndRunDnode();
}

1
source/dnode/mgmt/main/inc/dnd.h
View File

@ -136,6 +136,7 @@ typedef struct SDnode {
const char *dndNodeLogStr(ENodeType ntype);
const char *dndNodeProcStr(ENodeType ntype);
const char *dndEventStr(EDndEvent ev);
EDndStatus dndGetStatus(SDnode *pDnode);
void dndSetStatus(SDnode *pDnode, EDndStatus stat);
void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId);

6
source/dnode/mgmt/main/src/dndExec.c
View File

@ -150,7 +150,7 @@ static SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper) {
}
static int32_t dndRunInSingleProcess(SDnode *pDnode) {
dInfo("dnode start to run in single process");
dInfo("dnode run in single process");
for (ENodeType n = DNODE; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
@ -189,7 +189,7 @@ static int32_t dndRunInSingleProcess(SDnode *pDnode) {
}
static int32_t dndRunInParentProcess(SDnode *pDnode) {
dInfo("dnode start to run in parent process");
dInfo("dnode run in parent process");
SMgmtWrapper *pDWrapper = &pDnode->wrappers[DNODE];
if (dndOpenNode(pDWrapper) != 0) {
dError("node:%s, failed to start since %s", pDWrapper->name, terrstr());
@ -275,7 +275,7 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) {
}
static int32_t dndRunInChildProcess(SDnode *pDnode) {
dInfo("dnode start to run in child process");
dInfo("dnode run in child process");
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
SMsgCb msgCb = dndCreateMsgcb(pWrapper);

4
source/dnode/mgmt/main/src/dndFile.c
View File

@ -179,7 +179,7 @@ int32_t dndReadShmFile(SDnode *pDnode) {
}
}
if (!tsMultiProcess || pDnode->ntype == DNODE || pDnode->ntype == DNODE) {
if (!tsMultiProcess || pDnode->ntype == DNODE || pDnode->ntype == NODE_MAX) {
for (ENodeType ntype = DNODE; ntype < NODE_MAX; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
if (pWrapper->shm.id >= 0) {
@ -197,7 +197,7 @@ int32_t dndReadShmFile(SDnode *pDnode) {
dDebug("shmid:%d, is attached, size:%d", pWrapper->shm.id, pWrapper->shm.size);
}
dDebug("successed to open %s", file);
dDebug("successed to load %s", file);
code = 0;
_OVER:

2
source/dnode/mgmt/main/src/dndMsg.c
View File

@ -66,7 +66,7 @@ void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSet) {
dTrace("msg:%p, is created, handle:%p app:%p user:%s", pMsg, pRpc->handle, pRpc->ahandle, pMsg->user);
code = (*msgFp)(pWrapper, pMsg);
} else if (pWrapper->procType == PROC_PARENT) {
dTrace("msg:%p, is created and will put into child queue, handle:%p app:%p user:%s", pMsg, pRpc->handle,
dTrace("msg:%p, is created and put into child queue, handle:%p app:%p user:%s", pMsg, pRpc->handle,
pRpc->ahandle, pMsg->user);
code = taosProcPutToChildQ(pWrapper->pProc, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, PROC_REQ);
} else {

10
source/dnode/mgmt/main/src/dndObj.c
View File

@ -61,7 +61,7 @@ static void dndClearVars(SDnode *pDnode) {
taosMemoryFreeClear(pDnode->secondEp);
taosMemoryFreeClear(pDnode->dataDir);
taosMemoryFree(pDnode);
dDebug("dnode object memory is cleared, data:%p", pDnode);
dDebug("dnode memory is cleared, data:%p", pDnode);
}
SDnode *dndCreate(const SDnodeOpt *pOption) {
@ -117,14 +117,14 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
SMsgCb msgCb = dndCreateMsgcb(&pDnode->wrappers[0]);
tmsgSetDefaultMsgCb(&msgCb);
dInfo("dnode object is created, data:%p", pDnode);
dInfo("dnode is created, data:%p", pDnode);
code = 0;
_OVER:
if (code != 0 && pDnode) {
dndClearVars(pDnode);
pDnode = NULL;
dError("failed to create dnode object since %s", terrstr());
dError("failed to create dnode since %s", terrstr());
}
return pDnode;
@ -147,11 +147,11 @@ void dndClose(SDnode *pDnode) {
}
dndClearVars(pDnode);
dInfo("dnode object is closed, data:%p", pDnode);
dInfo("dnode is closed, data:%p", pDnode);
}
void dndHandleEvent(SDnode *pDnode, EDndEvent event) {
dInfo("dnode object receive event %d, data:%p", event, pDnode);
dInfo("dnode receive %s event, data:%p", dndEventStr(event), pDnode);
if (event == DND_EVENT_STOP) {
pDnode->event = event;
}

13
source/dnode/mgmt/main/src/dndStr.c
View File

@ -62,3 +62,16 @@ const char *dndNodeProcStr(ENodeType ntype) {
return "taosd";
}
}
const char *dndEventStr(EDndEvent ev) {
switch (ev) {
case DND_EVENT_START:
return "start";
case DND_EVENT_STOP:
return "stop";
case DND_EVENT_CHILD:
return "child";
default:
return "UNKNOWN";
}
}

4
source/os/src/osProc.c
View File

@ -39,7 +39,7 @@ void taosSetProcName(int32_t argc, char **argv, const char *name) {
argv[i][j] = 0;
}
if (i == 0) {
tstrncpy(argv[0], name, len);
tstrncpy(argv[0], name, len + 1);
}
}
}
@ -48,5 +48,5 @@ void taosSetProcPath(int32_t argc, char **argv) { tsProcPath = argv[0]; }
bool taosProcExists(int32_t pid) {
int32_t p = getpgid(pid);
return p == 0;
return p >= 0;
}

15
source/util/src/tprocess.c
View File

@ -336,7 +336,7 @@ SProcObj *taosProcInit(const SProcCfg *pCfg) {
pProc->parentConsumeFp = pCfg->parentConsumeFp;
pProc->isChild = pCfg->isChild;
uDebug("proc:%s, is initialized, child:%d child queue:%p parent queue:%p", pProc->name, pProc->isChild,
uDebug("proc:%s, is initialized, isChild:%d child queue:%p parent queue:%p", pProc->name, pProc->isChild,
pProc->pChildQueue, pProc->pParentQueue);
return pProc;
@ -370,7 +370,7 @@ static void taosProcThreadLoop(SProcObj *pProc) {
freeBodyFp = pProc->parentFreeBodyFp;
}
uDebug("proc:%s, start to get msg from queue:%p", pProc->name, pQueue);
uDebug("proc:%s, start to get msg from queue:%p, isChild:%d", pProc->name, pQueue, pProc->isChild);
while (1) {
int32_t numOfMsgs = taosProcQueuePop(pQueue, &pHead, &headLen, &pBody, &bodyLen, &ftype, mallocHeadFp, freeHeadFp,
@ -399,19 +399,19 @@ int32_t taosProcRun(SProcObj *pProc) {
return -1;
}
uDebug("proc:%s, start to consume queue:%p", pProc->name, pProc->pChildQueue);
uDebug("proc:%s, start to consume queue:%p, thread:%" PRId64, pProc->name, pProc->pChildQueue, pProc->thread);
return 0;
}
static void taosProcStop(SProcObj *pProc) {
if (!taosCheckPthreadValid(pProc->thread)) return;
uDebug("proc:%s, start to join thread", pProc->name);
uDebug("proc:%s, start to join thread:%" PRId64 ", isChild:%d", pProc->name, pProc->thread, pProc->isChild);
SProcQueue *pQueue;
if (pProc->isChild) {
pQueue = pProc->pParentQueue;
} else {
pQueue = pProc->pChildQueue;
} else {
pQueue = pProc->pParentQueue;
}
tsem_post(&pQueue->sem);
taosThreadJoin(pProc->thread, NULL);
@ -419,8 +419,9 @@ static void taosProcStop(SProcObj *pProc) {
void taosProcCleanup(SProcObj *pProc) {
if (pProc != NULL) {
uDebug("proc:%s, clean up", pProc->name);
uDebug("proc:%s, start to clean up", pProc->name);
taosProcStop(pProc);
uDebug("proc:%s, is cleaned up", pProc->name);
// taosProcCleanupQueue(pProc->pChildQueue);
// taosProcCleanupQueue(pProc->pParentQueue);
taosMemoryFree(pProc);