shm

include/os/osShm.h

@@ -22,11 +22,11 @@ extern "C" {
 
 typedef struct {
   int32_t id;
-  int32_t size;
+  int64_t size;
   void*   ptr;
 } SShm;
 
-int32_t taosCreateShm(SShm *pShm, int32_t shmsize) ;
+int32_t taosCreateShm(SShm *pShm, int64_t shmsize) ;
 void    taosDropShm(SShm *pShm);
 int32_t taosAttachShm(SShm *pShm);
 void    taosDetachShm(SShm *pShm);

source/dnode/mgmt/main/inc/dnd.h

@@ -128,7 +128,7 @@ typedef struct SDnode {
   EDndStatus   status;
   EDndEvent    event;
   SStartupReq  startup;
-  TdFilePtr    runtimeFile;
+  TdFilePtr    lockfile;
   STransMgmt   trans;
   SMgmtWrapper wrappers[NODE_MAX];
 } SDnode;

source/dnode/mgmt/main/inc/dndInt.h

@@ -54,9 +54,9 @@ int32_t dndInitMsgHandle(SDnode *pDnode);
 void    dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp);
 
 // dndFile.c
-int32_t dndOpenRuntimeFile(SDnode *pDnode);
-int32_t dndWriteRuntimeFile(SDnode *pDnode);
-void    dndCloseRuntimeFile(SDnode *pDnode);
+TdFilePtr dndCheckRunning(const char *dataDir);
+int32_t   dndReadShmFile(SDnode *pDnode);
+int32_t   dndWriteShmFile(SDnode *pDnode);
 
 #ifdef __cplusplus
 }

source/dnode/mgmt/main/src/dndExec.c

@@ -160,7 +160,12 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) {
       return -1;
     }
 
-    SProcCfg cfg = {.parentConsumeFp = (ProcConsumeFp)dndConsumeParentQueue,
+    SProcCfg cfg = {.childConsumeFp = (ProcConsumeFp)dndConsumeChildQueue,
+                    .childMallocHeadFp = (ProcMallocFp)taosAllocateQitem,
+                    .childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
+                    .childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
+                    .childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
+                    .parentConsumeFp = (ProcConsumeFp)dndConsumeParentQueue,
                     .parentdMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
                     .parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
                     .parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
@@ -176,7 +181,7 @@ static int32_t dndRunInParentProcess(SDnode *pDnode) {
     }
   }
 
-  if (dndWriteRuntimeFile(pDnode) != 0) {
+  if (dndWriteShmFile(pDnode) != 0) {
     dError("failed to write runtime file since %s", terrstr());
     return -1;
   }
@@ -220,6 +225,11 @@ static int32_t dndRunInChildProcess(SDnode *pDnode) {
                   .childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
                   .childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
                   .childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
+                  .parentConsumeFp = (ProcConsumeFp)dndConsumeParentQueue,
+                  .parentdMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
+                  .parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
+                  .parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
+                  .parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
                   .shm = pWrapper->shm,
                   .pParent = pWrapper,
                   .name = pWrapper->name};

source/dnode/mgmt/main/src/dndFile.c

@@ -117,7 +117,30 @@ _OVER:
   return code;
 }
 
-int32_t dndOpenRuntimeFile(SDnode *pDnode) {
+TdFilePtr dndCheckRunning(const char *dataDir) {
+  char filepath[PATH_MAX] = {0};
+  snprintf(filepath, sizeof(filepath), "%s%s.running", dataDir, TD_DIRSEP);
+
+  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;
+}
+
+int32_t dndReadShmFile(SDnode *pDnode) {
   int32_t   code = -1;
   char      itemName[24] = {0};
   char      content[MAXLEN + 1] = {0};
@@ -125,17 +148,11 @@ int32_t dndOpenRuntimeFile(SDnode *pDnode) {
   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);
+  snprintf(file, sizeof(file), "%s%s.shmfile", pDnode->dataDir, TD_DIRSEP);
+  pFile = taosOpenFile(file, TD_FILE_READ);
   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());
+    dDebug("file %s not exist", file);
+    code = 0;
     goto _OVER;
   }
 
@@ -150,14 +167,14 @@ int32_t dndOpenRuntimeFile(SDnode *pDnode) {
     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;
+      if (shmid && shmid->type == cJSON_String) {
+        pDnode->wrappers[ntype].shm.id = atoi(shmid->valuestring);
       }
 
       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 (shmsize && shmsize->type == cJSON_String) {
+        pDnode->wrappers[ntype].shm.size = atoll(shmsize->valuestring);
       }
     }
   }
@@ -166,7 +183,7 @@ int32_t dndOpenRuntimeFile(SDnode *pDnode) {
     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);
+        dDebug("shmid:%d, is closed, size:%" PRId64, pWrapper->shm.id, pWrapper->shm.size);
         taosDropShm(&pWrapper->shm);
       }
     }
@@ -177,7 +194,7 @@ int32_t dndOpenRuntimeFile(SDnode *pDnode) {
       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("shmid:%d, is attached, size:%" PRId64, pWrapper->shm.id, pWrapper->shm.size);
   }
 
   dDebug("successed to open %s", file);
@@ -185,16 +202,12 @@ int32_t dndOpenRuntimeFile(SDnode *pDnode) {
 
 _OVER:
   if (root != NULL) cJSON_Delete(root);
-  if (code != 0) {
-    if (pFile != NULL) taosCloseFile(&pFile);
-  } else {
-    pDnode->runtimeFile = pFile;
-  }
+  if (pFile != NULL) taosCloseFile(&pFile);
 
   return code;
 }
 
-int32_t dndWriteRuntimeFile(SDnode *pDnode) {
+int32_t dndWriteShmFile(SDnode *pDnode) {
   int32_t   code = -1;
   int32_t   len = 0;
   char      content[MAXLEN + 1] = {0};
@@ -202,8 +215,8 @@ int32_t dndWriteRuntimeFile(SDnode *pDnode) {
   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);
+  snprintf(file, sizeof(file), "%s%s.shmfile.bak", pDnode->dataDir, TD_DIRSEP);
+  snprintf(realfile, sizeof(realfile), "%s%s.shmfile", pDnode->dataDir, TD_DIRSEP);
 
   pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
   if (pFile == NULL) {
@@ -215,11 +228,13 @@ int32_t dndWriteRuntimeFile(SDnode *pDnode) {
   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);
+    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);
+      len += snprintf(content + len, MAXLEN - len, "  \"%s_shmsize\": \"%" PRId64 "\"\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, "  \"%s_shmsize\": \"%" PRId64 "\",\n", dndNodeProcStr(ntype),
+                      pWrapper->shm.size);
     }
   }
   len += snprintf(content + len, MAXLEN - len, "}\n");
@@ -254,11 +269,3 @@ _OVER:
 
   return code;
 }
-
-void dndCloseRuntimeFile(SDnode *pDnode) {
-  if (pDnode->runtimeFile) {
-    taosUnLockFile(pDnode->runtimeFile);
-    taosCloseFile(&pDnode->runtimeFile);
-    pDnode->runtimeFile = NULL;
-  }
-}
\ No newline at end of file

source/dnode/mgmt/main/src/dndObj.c

@@ -34,6 +34,12 @@ static int32_t dndInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
+
+  pDnode->lockfile = dndCheckRunning(pDnode->dataDir);
+  if (pDnode->lockfile == NULL) {
+    return -1;
+  }
+
   return 0;
 }
 
@@ -42,7 +48,11 @@ static void dndClearVars(SDnode *pDnode) {
     SMgmtWrapper *pMgmt = &pDnode->wrappers[n];
     taosMemoryFreeClear(pMgmt->path);
   }
-  dndCloseRuntimeFile(pDnode);
+  if (pDnode->lockfile != NULL) {
+    taosUnLockFile(pDnode->lockfile);
+    taosCloseFile(&pDnode->lockfile);
+    pDnode->lockfile = NULL;
+  }
   taosMemoryFreeClear(pDnode->localEp);
   taosMemoryFreeClear(pDnode->localFqdn);
   taosMemoryFreeClear(pDnode->firstEp);
@@ -96,8 +106,8 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
     goto _OVER;
   }
 
-  if (dndOpenRuntimeFile(pDnode) != 0) {
-    dError("failed to open runtime file since %s", terrstr());
+  if (dndReadShmFile(pDnode) != 0) {
+    dError("failed to read shm file since %s", terrstr());
     goto _OVER;
   }
 

source/os/src/osShm.c

@@ -17,8 +17,8 @@
 #define _DEFAULT_SOURCE
 #include "os.h"
 
-int32_t taosCreateShm(SShm* pShm, int32_t shmsize) {
-  int32_t shmid = shmget(IPC_PRIVATE, shmsize, IPC_CREAT | 0600);
+int32_t taosCreateShm(SShm* pShm, int64_t shmsize) {
+  int32_t shmid = shmget(IPC_PRIVATE, (size_t)shmsize, IPC_CREAT | 0600);
   if (shmid < 0) {
     return -1;
   }