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-30 14:06:36 +08:00
parent 6ed3c2490b
commit a341c476c0
8 changed files with 320 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/os/os.h
View File

@ -73,7 +73,6 @@ extern "C" {
#include <wchar.h> #include <wchar.h>
#include <wctype.h> #include <wctype.h>
#include "osAtomic.h" #include "osAtomic.h"
#include "osDef.h" #include "osDef.h"
#include "osDir.h" #include "osDir.h"
@ -87,6 +86,7 @@ extern "C" {
#include "osThread.h" #include "osThread.h"
#include "osSemaphore.h" #include "osSemaphore.h"
#include "osSignal.h" #include "osSignal.h"
#include "osShm.h"
#include "osSleep.h" #include "osSleep.h"
#include "osSocket.h" #include "osSocket.h"
#include "osString.h" #include "osString.h"

38
include/os/osShm.h Normal file
View File

@ -0,0 +1,38 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_OS_SHM_H_
#define _TD_OS_SHM_H_
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
int32_t id;
int32_t size;
void* ptr;
} SShm;
int32_t taosCreateShm(SShm *pShm, int32_t shmsize) ;
void taosDropShm(SShm *pShm);
int32_t taosAttachShm(SShm *pShm);
void taosDetachShm(SShm *pShm);
#ifdef __cplusplus
}
#endif
#endif /*_TD_OS_SHM_H_*/

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

@ -96,6 +96,7 @@ typedef struct SMgmtWrapper {
bool required; bool required;
EProcType procType; EProcType procType;
SProcObj *pProc; SProcObj *pProc;
SShm shm;
void *pMgmt; void *pMgmt;
SDnode *pDnode; SDnode *pDnode;
NodeMsgFp msgFps[TDMT_MAX]; NodeMsgFp msgFps[TDMT_MAX];
@ -127,7 +128,7 @@ typedef struct SDnode {
EDndStatus status; EDndStatus status;
EDndEvent event; EDndEvent event;
SStartupReq startup; SStartupReq startup;
TdFilePtr lockfile; TdFilePtr runtimeFile;
STransMgmt trans; STransMgmt trans;
SMgmtWrapper wrappers[NODE_MAX]; SMgmtWrapper wrappers[NODE_MAX];
} SDnode; } SDnode;

6
source/dnode/mgmt/main/inc/dndInt.h
View File

@ -34,7 +34,6 @@ int32_t dndInit();
void dndCleanup(); void dndCleanup();
const char *dndStatStr(EDndStatus stat); const char *dndStatStr(EDndStatus stat);
void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup); void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup);
TdFilePtr dndCheckRunning(const char *dataDir);
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg); void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg);
// dndMsg.c // dndMsg.c
@ -58,6 +57,11 @@ void dndCleanupClient(SDnode *pDnode);
int32_t dndInitMsgHandle(SDnode *pDnode); int32_t dndInitMsgHandle(SDnode *pDnode);
void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp); void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp);
// dndFile.c
int32_t dndOpenRuntimeFile(SDnode *pDnode);
int32_t dndWriteRuntimeFile(SDnode *pDnode);
void dndCloseRuntimeFile(SDnode *pDnode);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

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

@ -19,13 +19,12 @@
#define MAXLEN 1024 #define MAXLEN 1024
int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed) { int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed) {
int32_t code = TSDB_CODE_NODE_PARSE_FILE_ERROR; int32_t code = TSDB_CODE_NODE_PARSE_FILE_ERROR;
int32_t len = 0; int64_t len = 0;
const int32_t maxLen = MAXLEN; char content[MAXLEN + 1] = {0};
char content[MAXLEN + 1] = {0}; cJSON *root = NULL;
cJSON *root = NULL; char file[PATH_MAX];
char file[PATH_MAX]; TdFilePtr pFile = NULL;
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name); snprintf(file, sizeof(file), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
pFile = taosOpenFile(file, TD_FILE_READ); pFile = taosOpenFile(file, TD_FILE_READ);
@ -35,13 +34,12 @@ int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed) {
goto _OVER; goto _OVER;
} }
len = (int32_t)taosReadFile(pFile, content, maxLen); len = taosReadFile(pFile, content, MAXLEN);
if (len <= 0) { if (len <= 0) {
dError("failed to read %s since content is null", file); dError("failed to read %s since content is null", file);
goto _OVER; goto _OVER;
} }
content[len] = 0;
root = cJSON_Parse(content); root = cJSON_Parse(content);
if (root == NULL) { if (root == NULL) {
dError("failed to read %s since invalid json format", file); dError("failed to read %s since invalid json format", file);
@ -55,8 +53,8 @@ int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed) {
} }
*pDeployed = deployed->valueint != 0; *pDeployed = deployed->valueint != 0;
code = 0;
dDebug("succcessed to read file %s, deployed:%d", file, *pDeployed); dDebug("succcessed to read file %s, deployed:%d", file, *pDeployed);
code = 0;
_OVER: _OVER:
if (root != NULL) cJSON_Delete(root); if (root != NULL) cJSON_Delete(root);
@ -67,31 +65,41 @@ _OVER:
} }
int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed) { int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed) {
char file[PATH_MAX] = {0}; int32_t code = -1;
snprintf(file, sizeof(file), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name); int32_t len = 0;
char content[MAXLEN + 1] = {0};
char file[PATH_MAX] = {0};
char realfile[PATH_MAX] = {0};
TdFilePtr pFile = NULL;
TdFilePtr pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC); snprintf(file, sizeof(file), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
snprintf(realfile, sizeof(realfile), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) { if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to write %s since %s", file, terrstr()); dError("failed to write %s since %s", file, terrstr());
return -1; goto _OVER;
} }
int32_t len = 0; len += snprintf(content + len, MAXLEN - len, "{\n");
const int32_t maxLen = MAXLEN; len += snprintf(content + len, MAXLEN - len, " \"deployed\": %d\n", deployed);
char content[MAXLEN + 1] = {0}; len += snprintf(content + len, MAXLEN - len, "}\n");
len += snprintf(content + len, maxLen - len, "{\n"); if (taosWriteFile(pFile, content, len) != len) {
len += snprintf(content + len, maxLen - len, " \"deployed\": %d\n", deployed); terrno = TAOS_SYSTEM_ERROR(errno);
len += snprintf(content + len, maxLen - len, "}\n"); dError("failed to write file:%s since %s", file, terrstr());
goto _OVER;
}
if (taosFsyncFile(pFile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to fsync file:%s since %s", file, terrstr());
goto _OVER;
}
taosWriteFile(pFile, content, len);
taosFsyncFile(pFile);
taosCloseFile(&pFile); taosCloseFile(&pFile);
char realfile[PATH_MAX] = {0};
snprintf(realfile, sizeof(realfile), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
if (taosRenameFile(file, realfile) != 0) { if (taosRenameFile(file, realfile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to rename %s since %s", file, terrstr()); dError("failed to rename %s since %s", file, terrstr());
@ -99,5 +107,158 @@ int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed) {
} }
dInfo("successed to write %s, deployed:%d", realfile, deployed); dInfo("successed to write %s, deployed:%d", realfile, deployed);
return 0; code = 0;
_OVER:
if (pFile != NULL) {
taosCloseFile(&pFile);
}
return code;
} }
int32_t dndOpenRuntimeFile(SDnode *pDnode) {
int32_t code = -1;
char itemName[24] = {0};
char content[MAXLEN + 1] = {0};
char file[PATH_MAX] = {0};
cJSON *root = NULL;
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s.running", pDnode->dataDir, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to open file:%s since %s", file, terrstr());
goto _OVER;
}
if (taosLockFile(pFile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", file, terrstr());
goto _OVER;
}
if (taosReadFile(pFile, content, MAXLEN) > 0) {
root = cJSON_Parse(content);
if (root == NULL) {
terrno = TSDB_CODE_NODE_PARSE_FILE_ERROR;
dError("failed to read %s since invalid json format", file);
goto _OVER;
}
for (ENodeType ntype = DNODE + 1; ntype < NODE_MAX; ++ntype) {
snprintf(itemName, sizeof(itemName), "%s_shmid", dndNodeProcStr(ntype));
cJSON *shmid = cJSON_GetObjectItem(root, itemName);
if (shmid && shmid->type == cJSON_Number) {
pDnode->wrappers[ntype].shm.id = shmid->valueint;
}
snprintf(itemName, sizeof(itemName), "%s_shmsize", dndNodeProcStr(ntype));
cJSON *shmsize = cJSON_GetObjectItem(root, itemName);
if (shmsize && shmsize->type == cJSON_Number) {
pDnode->wrappers[ntype].shm.size = shmsize->valueint;
}
}
}
if (tsMultiProcess || pDnode->ntype == DNODE) {
for (ENodeType ntype = DNODE; ntype < NODE_MAX; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
if (pWrapper->shm.id > 0) {
dDebug("shmid:%d, is closed, size:%d", pWrapper->shm.id, pWrapper->shm.size);
taosDropShm(&pWrapper->shm);
}
}
} else {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
if (taosAttachShm(&pWrapper->shm) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("shmid:%d, failed to attach since %s", pWrapper->shm.id, terrstr());
goto _OVER;
}
dDebug("shmid:%d, is attached, size:%d", pWrapper->shm.id, pWrapper->shm.size);
}
dDebug("successed to open %s", file);
code = 0;
_OVER:
if (root != NULL) cJSON_Delete(root);
if (code != 0) {
if (pFile != NULL) taosCloseFile(&pFile);
} else {
pDnode->runtimeFile = pFile;
}
return code;
}
int32_t dndWriteRuntimeFile(SDnode *pDnode) {
int32_t code = -1;
int32_t len = 0;
char content[MAXLEN + 1] = {0};
char file[PATH_MAX] = {0};
char realfile[PATH_MAX] = {0};
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s.running.bak", pDnode->dataDir, TD_DIRSEP);
snprintf(realfile, sizeof(realfile), "%s%s.running", pDnode->dataDir, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to open file:%s since %s", file, terrstr());
goto _OVER;
}
len += snprintf(content + len, MAXLEN - len, "{\n");
for (ENodeType ntype = DNODE + 1; ntype < NODE_MAX; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
len += snprintf(content + len, MAXLEN - len, " \"%s_shmid\": %d,\n", dndNodeProcStr(ntype), pWrapper->shm.id);
if (ntype == NODE_MAX - 1) {
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\": %d\n", dndNodeProcStr(ntype), pWrapper->shm.size);
} else {
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\": %d,\n", dndNodeProcStr(ntype), pWrapper->shm.size);
}
}
len += snprintf(content + len, MAXLEN - len, "}\n");
if (taosWriteFile(pFile, content, len) != len) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to write file:%s since %s", file, terrstr());
goto _OVER;
}
if (taosFsyncFile(pFile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to fsync file:%s since %s", file, terrstr());
goto _OVER;
}
taosCloseFile(&pFile);
if (taosRenameFile(file, realfile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to rename %s to %s since %s", file, realfile, terrstr());
return -1;
}
dDebug("successed to write %s", realfile);
code = 0;
_OVER:
if (pFile != NULL) {
taosCloseFile(&pFile);
}
return code;
}
void dndCloseRuntimeFile(SDnode *pDnode) {
if (pDnode->runtimeFile) {
taosUnLockFile(pDnode->runtimeFile);
taosCloseFile(&pDnode->runtimeFile);
pDnode->runtimeFile = NULL;
}
}

23
source/dnode/mgmt/main/src/dndInt.c
View File

@ -84,29 +84,6 @@ void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup) {
pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING); pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING);
} }
TdFilePtr dndCheckRunning(const char *dataDir) {
char filepath[PATH_MAX] = {0};
snprintf(filepath, sizeof(filepath), "%s/.running", dataDir);
TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", filepath, terrstr());
return NULL;
}
int32_t ret = taosLockFile(pFile);
if (ret != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", filepath, terrstr());
taosCloseFile(&pFile);
return NULL;
}
dDebug("file:%s is locked", filepath);
return pFile;
}
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) { void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("startup req is received"); dDebug("startup req is received");
SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq)); SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq));

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

@ -16,7 +16,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "dndInt.h" #include "dndInt.h"
static int32_t dndInitMemory(SDnode *pDnode, const SDnodeOpt *pOption) { static int32_t dndInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
pDnode->numOfSupportVnodes = pOption->numOfSupportVnodes; pDnode->numOfSupportVnodes = pOption->numOfSupportVnodes;
pDnode->serverPort = pOption->serverPort; pDnode->serverPort = pOption->serverPort;
pDnode->dataDir = strdup(pOption->dataDir); pDnode->dataDir = strdup(pOption->dataDir);
@ -37,16 +37,12 @@ static int32_t dndInitMemory(SDnode *pDnode, const SDnodeOpt *pOption) {
return 0; return 0;
} }
static void dndClearMemory(SDnode *pDnode) { static void dndClearVars(SDnode *pDnode) {
for (ENodeType n = 0; n < NODE_MAX; ++n) { for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pMgmt = &pDnode->wrappers[n]; SMgmtWrapper *pMgmt = &pDnode->wrappers[n];
taosMemoryFreeClear(pMgmt->path); taosMemoryFreeClear(pMgmt->path);
} }
if (pDnode->lockfile != NULL) { dndCloseRuntimeFile(pDnode);
taosUnLockFile(pDnode->lockfile);
taosCloseFile(&pDnode->lockfile);
pDnode->lockfile = NULL;
}
taosMemoryFreeClear(pDnode->localEp); taosMemoryFreeClear(pDnode->localEp);
taosMemoryFreeClear(pDnode->localFqdn); taosMemoryFreeClear(pDnode->localFqdn);
taosMemoryFreeClear(pDnode->firstEp); taosMemoryFreeClear(pDnode->firstEp);
@ -68,13 +64,21 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
goto _OVER; goto _OVER;
} }
if (dndInitMemory(pDnode, pOption) != 0) { if (dndInitVars(pDnode, pOption) != 0) {
dError("failed to init variables since %s", terrstr());
goto _OVER; goto _OVER;
} }
dndSetStatus(pDnode, DND_STAT_INIT); dndSetStatus(pDnode, DND_STAT_INIT);
pDnode->lockfile = dndCheckRunning(pDnode->dataDir); dmGetMgmtFp(&pDnode->wrappers[DNODE]);
if (pDnode->lockfile == NULL) { mmGetMgmtFp(&pDnode->wrappers[MNODE]);
vmGetMgmtFp(&pDnode->wrappers[VNODES]);
qmGetMgmtFp(&pDnode->wrappers[QNODE]);
smGetMgmtFp(&pDnode->wrappers[SNODE]);
bmGetMgmtFp(&pDnode->wrappers[BNODE]);
if (dndOpenRuntimeFile(pDnode) != 0) {
dError("failed to open runtime file since %s", terrstr());
goto _OVER; goto _OVER;
} }
@ -88,13 +92,6 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
goto _OVER; goto _OVER;
} }
dmGetMgmtFp(&pDnode->wrappers[DNODE]);
mmGetMgmtFp(&pDnode->wrappers[MNODE]);
vmGetMgmtFp(&pDnode->wrappers[VNODES]);
qmGetMgmtFp(&pDnode->wrappers[QNODE]);
smGetMgmtFp(&pDnode->wrappers[SNODE]);
bmGetMgmtFp(&pDnode->wrappers[BNODE]);
if (dndInitMsgHandle(pDnode) != 0) { if (dndInitMsgHandle(pDnode) != 0) {
goto _OVER; goto _OVER;
} }
@ -117,7 +114,7 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
_OVER: _OVER:
if (code != 0 && pDnode) { if (code != 0 && pDnode) {
dndClearMemory(pDnode); dndClearVars(pDnode);
pDnode = NULL; pDnode = NULL;
dError("failed to create dnode object since %s", terrstr()); dError("failed to create dnode object since %s", terrstr());
} else { } else {
@ -146,7 +143,7 @@ void dndClose(SDnode *pDnode) {
dndCloseNode(pWrapper); dndCloseNode(pWrapper);
} }
dndClearMemory(pDnode); dndClearVars(pDnode);
dInfo("dnode object is closed, data:%p", pDnode); dInfo("dnode object is closed, data:%p", pDnode);
} }

71
source/os/src/osShm.c Normal file
View File

@ -0,0 +1,71 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define ALLOW_FORBID_FUNC
#define _DEFAULT_SOURCE
#include "os.h"
int32_t taosCreateShm(SShm* pShm, int32_t shmsize) {
int32_t shmid = shmget(IPC_PRIVATE, shmsize, IPC_CREAT | 0600);
if (shmid < 0) {
return -1;
}
void* shmptr = shmat(shmid, NULL, 0);
if (shmptr == NULL) {
return -1;
}
pShm->id = shmid;
pShm->size = shmsize;
pShm->ptr = shmptr;
return 0;
}
void taosDropShm(SShm* pShm) {
if (pShm->id > 0) {
if (pShm->ptr != NULL) {
shmdt(pShm->ptr);
}
shmctl(pShm->id, IPC_RMID, NULL);
}
pShm->id = 0;
pShm->size = 0;
pShm->ptr = NULL;
}
int32_t taosAttachShm(SShm* pShm) {
if (pShm->id > 0 && pShm->size > 0) {
pShm->ptr = shmat(pShm->id, NULL, 0);
if (pShm->ptr != NULL) {
return 0;
}
}
return -1;
}
void taosDetachShm(SShm* pShm) {
if (pShm->id > 0) {
if (pShm->ptr != NULL) {
shmdt(pShm->ptr);
pShm->ptr = NULL;
}
}
pShm->id = 0;
pShm->size = 0;
pShm->ptr = NULL;
}