add queue api

demo/insert.c

@@ -25,14 +25,14 @@ long    lInsertUserInfo()
         return RC_FAIL;
     }
  
-    memset(&stUser, 0, sizeof(dbUser));
+    conditinit(pstSavm, stUser, TBL_USER_INFO);          // 绑定变量
+
 //    stUser.acct_id = 10021;                              // 对结构体赋值
     strcpy(stUser.user_no,    "20180223");               // 对结构体赋值
     strcpy(stUser.user_type,  "1");                      // 对结构体赋值
     strcpy(stUser.user_nm,    "Savens Liu");             // 对结构体赋值
     strcpy(stUser.user_addr,  "China");                  // 对结构体赋值
     strcpy(stUser.user_phone, "18672911111");            // 对结构体赋值
-    defineinit(pstSavm, stUser, TBL_USER_INFO);          // 绑定变量
     decorate(pstSavm, dbUser, acct_id, FIELD_INCR);      // 自增llong型
  
     if(RC_SUCC != lInsert(pstSavm))      // 插入记录  

include/tvm.h

@@ -332,17 +332,18 @@ typedef long                 CREATE;
 #define defineinit(p,s,t)       do{ \
                                    p->stCond.uFldcmp = 0; \
                                    p->stUpdt.uFldcmp = 0; \
-                                   p->lSize = sizeof(s); \
+                                   p->lFind = 0;  \
                                    p->tblName = t; \
+                                   p->lSize = sizeof(s); \
                                    p->pstVoid = (void *)&(s);  \
                                 }while(0);
 
 #define conditinit(p,s,t)       do{ \
                                    p->stCond.uFldcmp = 0; \
                                    p->stUpdt.uFldcmp = 0; \
-                                   p->lSize = sizeof(s); \
-                                   p->tblName = t; \
                                    p->lFind = 0;  \
+                                   p->tblName = t; \
+                                   p->lSize = sizeof(s); \
                                    memset(&(s), 0, p->lSize); \
                                    p->pstVoid = (void *)&(s);  \
                                 }while(0);
@@ -356,6 +357,21 @@ typedef long                 CREATE;
                                    p->pstVoid = NULL;  \
                                 }while(0);
 
+#define queueinit(p,v,t)       do{ \
+                                   p->lSize = sizeof(v); \
+                                   p->tblName = t; \
+                                   memset(&(v), 0, p->lSize); \
+                                   p->pstVoid = (void *)&v;  \
+                                }while(0);
+
+#define queuerset(p,v,l,t)     do{ \
+                                   p->lSize = l; \
+                                   p->tblName = t; \
+                                   p->pstVoid = (void *)v;  \
+                                }while(0);
+
+#define queuebind(p,v,l,t)      queuerset(p,v,sizeof(l),t)
+
 #define stringsetv(p,s,f,...)   vSetCodField(&p->stCond, sizeof((s).f), (char *)(s).f - (char *)&(s)); \
                                 snprintf((s).f, sizeof((s).f), __VA_ARGS__);
 
@@ -675,9 +691,6 @@ extern "C" {
 extern    char*    sGetLog();
 extern    char*    sGetNode();
 extern    void*    pGetBoot();
-extern    void*    pGetSATvm();
-extern    void*    pCloneSATvm();
-extern    void     vCloneFree(SATvm *pstSavm);
 extern    long     lDefaultBoot();
 extern    TBoot*   pBootInitial();
 extern    size_t   lGetTblRow(TABLE t);
@@ -730,6 +743,16 @@ extern    long     lGetTblIndex(SATvm *pstSavm, char *pszTable, char *pszPart, T
 extern    long    lUnmakeConfig(char *pszFile);
 extern    long    lMakeConfig(char *pszFile);
 
+/*************************************************************************************************
+    stvm handle
+ *************************************************************************************************/
+extern    void*    pGetSATvm();
+
+// ptrhead use
+extern    void*    pCloneSATvm();
+extern    void     vCloneFree(SATvm *pstSavm);
+extern    void     vCloneQueue(SATvm *pstSovm, TABLE t);
+
 /*************************************************************************************************
     IPC Message and semaphore
  *************************************************************************************************/

src/tree.c

@@ -455,6 +455,34 @@ char*    pGetIndex(FdCond *pstCond, long lIdx, TblKey *pstKey, void *psvData, ch
     return pszIdx;
 }
 
+/*************************************************************************************************
+    description：set number of current attaches
+    parameters:
+        pstSavm                    --stvm handle
+        t                          --table
+    return:
+        void*                      --success
+ *************************************************************************************************/
+void    vSetQueueAttch(RunTime *pstRun, long lRevise)
+{
+    struct shmid_ds ds;
+
+    if(TYPE_MQUEUE == pstRun->m_lType || pstRun->m_pvAddr) 
+        return ;
+
+    memset(&ds, 0, sizeof(struct shmid_ds));
+    if(0 != shmctl(pstRun->m_shmID, IPC_STAT, &ds))
+    {
+        ds.shm_nattch = 1;
+        return ;
+    }
+
+    ds.shm_nattch = (ds.shm_nattch + lRevise) < 0 ? 1 : (ds.shm_nattch + lRevise);
+    ((TblDef *)pstRun->m_pvAddr)->m_lGroup = ds.shm_nattch;
+
+    return ;
+}
+
 /*************************************************************************************************
     description：get stvm handle
     parameters:
@@ -466,6 +494,23 @@ void*   pGetSATvm()
     return &g_stSavm;
 }
 
+/*************************************************************************************************
+    description：clone stvm handle by thread
+    parameters:
+    return:
+       void*
+  *************************************************************************************************/
+void    vCloneQueue(SATvm *pstSovm, TABLE t)
+{
+    RunTime  *pstRun = NULL;
+
+    if(!pstSovm || NULL == (pstRun = (RunTime *)pGetRunTime(pstSovm, t)))
+        return ;
+
+    vSetQueueAttch(pstRun, 1);
+    return ;
+}
+
 /*************************************************************************************************
     description：clone stvm handle by thread
     parameters:
@@ -1103,8 +1148,7 @@ void    vHoldRelease(SATvm *pstSavm)
 
         if(pstRun->m_pvAddr)
         {
-            if(TYPE_MQUEUE == pstRun->m_lType && ((TblDef *)pstRun->m_pvAddr)->m_lGroup > 0)
+            vSetQueueAttch(pstRun, -1);
-                ((TblDef *)pstRun->m_pvAddr)->m_lGroup --; // process exit
             shmdt(pstRun->m_pvAddr);
         }
         pstRun->m_pvAddr = NULL;
@@ -1141,8 +1185,7 @@ void    _vTblRelease(SATvm *pstSavm, TABLE t, bool bHold)
 
     if(pstRun->m_pvAddr)
     {
-        if(TYPE_MQUEUE == pstRun->m_lType && ((TblDef *)pstRun->m_pvAddr)->m_lGroup > 0)
+        vSetQueueAttch(pstRun, -1);
-            ((TblDef *)pstRun->m_pvAddr)->m_lGroup --; // process exit
         shmdt(pstRun->m_pvAddr);
     }
     pstRun->m_pvAddr = NULL;
@@ -1848,8 +1891,7 @@ void*    pInitMemTable(SATvm *pstSavm, TABLE t)
     }
 
     pstRun->m_bAttch = true;
-    if(TYPE_MQUEUE == pstRun->m_lType)
+    vSetQueueAttch(pstRun, 0);
-        ((TblDef *)pstRun->m_pvAddr)->m_lGroup ++; // process join 
     memcpy((void *)pGetTblDef(t), pstRun->m_pvAddr, sizeof(TblDef));
 
     if(pstSavm->lSize != lGetRowSize(t))
@@ -2107,7 +2149,7 @@ long    lInsertField(SATvm *pstSavm, TABLE t)
     if(NULL == (pstSavm = (SATvm *)pInitSATvm(SYS_TVM_FIELD)))
         return RC_FAIL;
 
-    defineinit(pstSavm, stField, SYS_TVM_FIELD)
+    conditbind(pstSavm, stField, SYS_TVM_FIELD)
     for(i = 0; i < lIdx; i ++)
     {
         memset(&stField, 0, sizeof(TField));
@@ -8490,7 +8532,7 @@ long    lInitDomain(SATvm *pstSavm)
     if(RC_SUCC != lGetDomainIndex(pstSavm, &lCount, &pstIndex))
         return RC_FAIL;
 
-    defineinit(pstSavm, pstIndex[i], SYS_TVM_INDEX)
+    conditbind(pstSavm, pstIndex[i], SYS_TVM_INDEX)
     for(i = 0; i < lCount; i ++)
     {
         pstIndex[i].m_lValid = 0;
@@ -8517,7 +8559,7 @@ long    lInitDomain(SATvm *pstSavm)
     {
         pstDomain[i].m_lStatus = RESOURCE_STOP;
 
-        defineinit(pstSavm, pstDomain[i], SYS_TVM_DOMAIN)
+        conditbind(pstSavm, pstDomain[i], SYS_TVM_DOMAIN)
         if(RC_SUCC != lInsert(pstSavm))
         {
             TFree(pstDomain);
@@ -8610,7 +8652,7 @@ long    lStartupTvm(TBoot *pstBoot)
         return RC_FAIL;
     */
 
-    defineinit(pstSavm, stIndex, SYS_TVM_INDEX)
+    conditbind(pstSavm, stIndex, SYS_TVM_INDEX)
     if(RC_SUCC != lInsert(pstSavm))
         return RC_FAIL;
 
@@ -8635,7 +8677,7 @@ long    lStartupTvm(TBoot *pstBoot)
     strncpy(stIndex.m_szPart, pstBoot->m_szNode, sizeof(stIndex.m_szPart));
     strncpy(stIndex.m_szTable, sGetTableName(SYS_TVM_FIELD), sizeof(stIndex.m_szTable));
     
-    defineinit(pstSavm, stIndex, SYS_TVM_INDEX)
+    conditbind(pstSavm, stIndex, SYS_TVM_INDEX)
     if(RC_SUCC != lInsert(pstSavm))
         return RC_FAIL;
 
@@ -8663,7 +8705,7 @@ long    lStartupTvm(TBoot *pstBoot)
     strncpy(stIndex.m_szPart, pstBoot->m_szNode, sizeof(stIndex.m_szPart));
     strncpy(stIndex.m_szTable, sGetTableName(SYS_TVM_DOMAIN), sizeof(stIndex.m_szTable));
     
-    defineinit(pstSavm, stIndex, SYS_TVM_INDEX)
+    conditbind(pstSavm, stIndex, SYS_TVM_INDEX)
     if(RC_SUCC != lInsert(pstSavm))
         return RC_FAIL;
 
@@ -8691,7 +8733,7 @@ long    lStartupTvm(TBoot *pstBoot)
     strncpy(stIndex.m_szPart, pstBoot->m_szNode, sizeof(stIndex.m_szPart));
     strncpy(stIndex.m_szTable, sGetTableName(SYS_TVM_SEQUE), sizeof(stIndex.m_szTable));
     
-    defineinit(pstSavm, stIndex, SYS_TVM_INDEX)
+    conditbind(pstSavm, stIndex, SYS_TVM_INDEX)
     if(RC_SUCC != lInsert(pstSavm))
         return RC_FAIL;