stmt reqid incr one

source/libs/transport/test/svrBench.c

@@ -26,21 +26,40 @@ TdFilePtr   pDataFile = NULL;
 STaosQueue *qhandle = NULL;
 STaosQset  *qset = NULL;
 
+int32_t balance = 0;
+
+typedef struct {
+  int32_t      numOfThread;
+  STaosQueue **qhandle;
+  STaosQset  **qset;
+
+} MultiThreadQhandle;
+
+typedef struct TThread {
+  TdThread thread;
+  int      idx;
+} TThread;
+
+MultiThreadQhandle *multiQ = NULL;
+
 void initLogEnv() {
-  const char  *logDir =  "/tmp/trans_svr"; 
-  const char*  defaultLogFileNamePrefix = "taoslog";
+  const char   *logDir = "/tmp/trans_svr";
+  const char   *defaultLogFileNamePrefix = "taoslog";
   const int32_t maxLogFileNum = 10000;
   tsAsyncLog = 0;
-  //idxDebugFlag = 143;
+  // idxDebugFlag = 143;
   strcpy(tsLogDir, logDir);
   taosRemoveDir(tsLogDir);
-  taosMkDir(tsLogDir); 
-     
+  taosMkDir(tsLogDir);
+
   if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) {
-   printf("failed to open log file in directory:%s\n", tsLogDir);  
+    printf("failed to open log file in directory:%s\n", tsLogDir);
   }
 }
-void processShellMsg() {
+void *processShellMsg(void *arg) {
+  TThread *thread = (TThread *)arg;
+
+  int32_t    idx = thread->idx;
   static int num = 0;
   STaosQall *qall;
   SRpcMsg   *pRpcMsg, rpcMsg;
@@ -50,7 +69,7 @@ void processShellMsg() {
   qall = taosAllocateQall();
 
   while (1) {
-    int numOfMsgs = taosReadAllQitemsFromQset(qset, qall, &qinfo);
+    int numOfMsgs = taosReadAllQitemsFromQset(multiQ->qset[idx], qall, &qinfo);
     tDebug("%d shell msgs are received", numOfMsgs);
     if (numOfMsgs <= 0) break;
 
@@ -103,6 +122,7 @@ void processShellMsg() {
   }
 
   taosFreeQall(qall);
+  return NULL;
 }
 
 void processRequestMsg(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
@@ -111,8 +131,11 @@ void processRequestMsg(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
   pTemp = taosAllocateQitem(sizeof(SRpcMsg), DEF_QITEM);
   memcpy(pTemp, pMsg, sizeof(SRpcMsg));
 
+  int32_t idx = balance % multiQ->numOfThread;
   tDebug("request is received, type:%d, contLen:%d, item:%p", pMsg->msgType, pMsg->contLen, pTemp);
-  taosWriteQitem(qhandle, pTemp);
+  taosWriteQitem(multiQ->qhandle[idx], pTemp);
+  balance++;
+  if (balance >= multiQ->numOfThread) balance = 0;
 }
 
 int main(int argc, char *argv[]) {
@@ -162,7 +185,7 @@ int main(int argc, char *argv[]) {
   }
 
   rpcInit.connType = TAOS_CONN_SERVER;
-  
+
   initLogEnv();
 
   void *pRpc = rpcOpen(&rpcInit);
@@ -178,16 +201,35 @@ int main(int argc, char *argv[]) {
     pDataFile = taosOpenFile(dataName, TD_FILE_APPEND | TD_FILE_CREATE | TD_FILE_WRITE);
     if (pDataFile == NULL) tInfo("failed to open data file, reason:%s", strerror(errno));
   }
-  qhandle = taosOpenQueue();
-  qset = taosOpenQset();
-  taosAddIntoQset(qset, qhandle, NULL);
 
-  processShellMsg();
+  int32_t numOfAthread = 5;
+  multiQ = taosMemoryMalloc(sizeof(numOfAthread));
+  multiQ->numOfThread = numOfAthread;
+  multiQ->qhandle = (STaosQueue **)taosMemoryMalloc(sizeof(STaosQueue *) * numOfAthread);
+  multiQ->qset = (STaosQset **)taosMemoryMalloc(sizeof(STaosQset *) * numOfAthread);
+
+  for (int i = 0; i < numOfAthread; i++) {
+    multiQ->qhandle[i] = taosOpenQueue();
+    multiQ->qset[i] = taosOpenQset();
+    taosAddIntoQset(multiQ->qset[i], multiQ->qhandle[i], NULL);
+  }
+  TThread *threads = taosMemoryMalloc(sizeof(TThread) * numOfAthread);
+  for (int i = 0; i < numOfAthread; i++) {
+    threads[i].idx = i;
+    taosThreadCreate(&(threads[i].thread), NULL, processShellMsg, (void *)&threads[i]);
+  }
+  // qhandle = taosOpenQueue();
+  // qset = taosOpenQset();
+  // taosAddIntoQset(qset, qhandle, NULL);
+
+  // processShellMsg();
 
   if (pDataFile != NULL) {
     taosCloseFile(&pDataFile);
     taosRemoveFile(dataName);
   }
+  int ch = getchar();
+  UNUSED(ch);
 
   return 0;
 }