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

fix start/stop/restart udfd

This commit is contained in:
slzhou 2022-04-22 15:21:26 +08:00
parent d70c74d44e
commit a395b21578
3 changed files with 25 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
source/dnode/mgmt/implement/src/dmHandle.c
View File

@ -220,13 +220,11 @@ static int32_t dmSpawnUdfd(SDnode *pDnode);
void dmUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal) { void dmUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal) {
dInfo("udfd process exited with status %" PRId64 ", signal %d", exitStatus, termSignal); dInfo("udfd process exited with status %" PRId64 ", signal %d", exitStatus, termSignal);
uv_close((uv_handle_t*)process, NULL);
SDnode *pDnode = process->data; SDnode *pDnode = process->data;
SUdfdData *pData = &pDnode->udfdData; if (exitStatus == 0 && termSignal == 0) {
if (atomic_load_8(&pData->stopping) != 0) { dInfo("udfd process exit due to SIGINT");
dDebug("udfd process exit due to stopping");
} else { } else {
uv_close((uv_handle_t*)&pData->ctrlPipe, NULL); dInfo("udfd process restart");
dmSpawnUdfd(pDnode); dmSpawnUdfd(pDnode);
} }
} }
@ -248,6 +246,7 @@ static int32_t dmSpawnUdfd(SDnode *pDnode) {
options.file = path; options.file = path;
options.exit_cb = dmUdfdExit; options.exit_cb = dmUdfdExit;
SUdfdData *pData = &pDnode->udfdData; SUdfdData *pData = &pDnode->udfdData;
uv_pipe_init(&pData->loop, &pData->ctrlPipe, 1); uv_pipe_init(&pData->loop, &pData->ctrlPipe, 1);
@ -260,6 +259,8 @@ static int32_t dmSpawnUdfd(SDnode *pDnode) {
options.stdio_count = 3; options.stdio_count = 3;
options.stdio = child_stdio; options.stdio = child_stdio;
options.flags = UV_PROCESS_DETACHED;
char dnodeIdEnvItem[32] = {0}; char dnodeIdEnvItem[32] = {0};
char thrdPoolSizeEnvItem[32] = {0}; char thrdPoolSizeEnvItem[32] = {0};
snprintf(dnodeIdEnvItem, 32, "%s=%d", "DNODE_ID", pDnode->data.dnodeId); snprintf(dnodeIdEnvItem, 32, "%s=%d", "DNODE_ID", pDnode->data.dnodeId);
@ -284,24 +285,31 @@ static void dmUdfdCloseWalkCb(uv_handle_t* handle, void* arg) {
} }
} }
void dmWatchUdfd(void *args) { static void dmUdfdStopAsyncCb(uv_async_t *async) {
SDnode *pDnode = async->data;
SUdfdData *pData = &pDnode->udfdData;
uv_stop(&pData->loop);
}
static void dmWatchUdfd(void *args) {
SDnode *pDnode = args; SDnode *pDnode = args;
SUdfdData *pData = &pDnode->udfdData; SUdfdData *pData = &pDnode->udfdData;
uv_loop_init(&pData->loop); uv_loop_init(&pData->loop);
uv_async_init(&pData->loop, &pData->stopAsync, dmUdfdStopAsyncCb);
pData->stopAsync.data = pDnode;
int32_t err = dmSpawnUdfd(pDnode); int32_t err = dmSpawnUdfd(pDnode);
atomic_store_32(&pData->spawnErr, err); atomic_store_32(&pData->spawnErr, err);
uv_barrier_wait(&pData->barrier); uv_barrier_wait(&pData->barrier);
uv_run(&pData->loop, UV_RUN_DEFAULT); uv_run(&pData->loop, UV_RUN_DEFAULT);
err = uv_loop_close(&pData->loop); uv_loop_close(&pData->loop);
while (err == UV_EBUSY) {
uv_walk(&pData->loop, dmUdfdCloseWalkCb, NULL); uv_walk(&pData->loop, dmUdfdCloseWalkCb, NULL);
uv_run(&pData->loop, UV_RUN_DEFAULT); uv_run(&pData->loop, UV_RUN_DEFAULT);
err = uv_loop_close(&pData->loop); uv_loop_close(&pData->loop);
}
return; return;
} }
int32_t dmStartUdfd(SDnode *pDnode) { static int32_t dmStartUdfd(SDnode *pDnode) {
SUdfdData *pData = &pDnode->udfdData; SUdfdData *pData = &pDnode->udfdData;
if (pData->startCalled) { if (pData->startCalled) {
dInfo("dnode-mgmt start udfd already called"); dInfo("dnode-mgmt start udfd already called");
@ -309,30 +317,27 @@ int32_t dmStartUdfd(SDnode *pDnode) {
} }
pData->startCalled = true; pData->startCalled = true;
uv_barrier_init(&pData->barrier, 2); uv_barrier_init(&pData->barrier, 2);
pData->stopping = 0;
uv_thread_create(&pData->thread, dmWatchUdfd, pDnode); uv_thread_create(&pData->thread, dmWatchUdfd, pDnode);
uv_barrier_wait(&pData->barrier); uv_barrier_wait(&pData->barrier);
pData->needCleanUp = true; pData->needCleanUp = true;
return pData->spawnErr; return pData->spawnErr;
} }
int32_t dmStopUdfd(SDnode *pDnode) { static int32_t dmStopUdfd(SDnode *pDnode) {
dInfo("dnode-mgmt to stop udfd. need cleanup: %d, spawn err: %d", dInfo("dnode-mgmt to stop udfd. need cleanup: %d, spawn err: %d",
pDnode->udfdData.needCleanUp, pDnode->udfdData.spawnErr); pDnode->udfdData.needCleanUp, pDnode->udfdData.spawnErr);
SUdfdData *pData = &pDnode->udfdData; SUdfdData *pData = &pDnode->udfdData;
if (!pData->needCleanUp) { if (!pData->needCleanUp) {
return 0; return 0;
} }
atomic_store_8(&pData->stopping, 1);
uv_barrier_destroy(&pData->barrier); uv_barrier_destroy(&pData->barrier);
if (pData->spawnErr == 0) { if (pData->spawnErr == 0) {
uv_process_kill(&pData->process, SIGINT); uv_process_kill(&pData->process, SIGINT);
} }
uv_stop(&pData->loop); uv_async_send(&pData->stopAsync);
uv_thread_join(&pData->thread); uv_thread_join(&pData->thread);
atomic_store_8(&pData->stopping, 0);
return 0; return 0;
} }

2
source/dnode/mgmt/interface/inc/dmDef.h
View File

@ -151,8 +151,8 @@ typedef struct SUdfdData {
uv_barrier_t barrier; uv_barrier_t barrier;
uv_process_t process; uv_process_t process;
int spawnErr; int spawnErr;
int8_t stopping;
uv_pipe_t ctrlPipe; uv_pipe_t ctrlPipe;
uv_async_t stopAsync;
} SUdfdData; } SUdfdData;
typedef struct SDnode { typedef struct SDnode {

2
source/libs/function/src/udfd.c
View File

@ -586,7 +586,7 @@ static int32_t udfdRun() {
udfdCloseClientRpc(); udfdCloseClientRpc();
uv_mutex_destroy(&global.udfsMutex); uv_mutex_destroy(&global.udfsMutex);
taosHashCleanup(global.udfsHash); taosHashCleanup(global.udfsHash);
return code; return 0;
} }
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {