Merge remote-tracking branch 'origin/3.0' into feature/config

include/libs/transport/trpc.h

@@ -78,6 +78,9 @@ typedef struct SRpcInit {
   // call back to retrieve the client auth info, for server app only
   int (*afp)(void *parent, char *tableId, char *spi, char *encrypt, char *secret, char *ckey);
 
+  // call back to keep conn or not
+  bool (*pfp)(void *parent, tmsg_t msgType);
+
   void *parent;
 } SRpcInit;
 

source/client/inc/clientInt.h

@@ -20,17 +20,17 @@
 extern "C" {
 #endif
 
-#include "taos.h"
 #include "common.h"
-#include "tmsg.h"
+#include "parser.h"
+#include "query.h"
+#include "taos.h"
 #include "tdef.h"
 #include "tep.h"
 #include "thash.h"
 #include "tlist.h"
+#include "tmsg.h"
 #include "tmsgtype.h"
 #include "trpc.h"
-#include "query.h"
-#include "parser.h"
 
 #include "config.h"
 
@@ -68,12 +68,10 @@ typedef struct SAppHbMgr {
   SHashObj* connInfo;    // hash<SClientHbKey, SHbConnInfo>
 } SAppHbMgr;
 
-
 typedef int32_t (*FHbRspHandle)(struct SAppHbMgr* pAppHbMgr, SClientHbRsp* pRsp);
 
 typedef int32_t (*FHbReqHandle)(SClientHbKey* connKey, void* param, SClientHbReq* req);
 
-
 typedef struct SClientHbMgr {
   int8_t inited;
   // ctl
@@ -85,7 +83,6 @@ typedef struct SClientHbMgr {
   FHbRspHandle    rspHandle[HEARTBEAT_TYPE_MAX];
 } SClientHbMgr;
 
-
 typedef struct SQueryExecMetric {
   int64_t start;   // start timestamp
   int64_t parsed;  // start to parse
@@ -221,17 +218,18 @@ int  taos_options_imp(TSDB_OPTION option, const char *str);
 
 void* openTransporter(const char* user, const char* auth, int32_t numOfThreads);
 
+bool persistConnForSpecificMsg(void* parenct, tmsg_t msgType);
 void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet);
 
 void initMsgHandleFp();
 
-TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass, const char *auth, const char *db, uint16_t port);
+TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
+                            uint16_t port);
 
 void* doFetchRow(SRequestObj* pRequest);
 
 void setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows);
 
-
 int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj** pRequest);
 
 int32_t parseSql(SRequestObj* pRequest, SQueryNode** pQuery);
@@ -255,6 +253,7 @@ int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* v
 // --- mq
 void hbMgrInitMqHbRspHandle();
 
+
 // config
 int32_t tscInitLog(const char *cfgDir, const char *envFile, const char *apolloUrl);
 int32_t tscInitCfg(const char *cfgDir, const char *envFile, const char *apolloUrl);

source/client/src/clientEnv.c

@@ -13,15 +13,15 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "os.h"
 #include "catalog.h"
 #include "clientInt.h"
 #include "clientLog.h"
+#include "os.h"
 #include "query.h"
 #include "scheduler.h"
-#include "tmsg.h"
 #include "tcache.h"
 #include "tglobal.h"
+#include "tmsg.h"
 #include "tref.h"
 #include "trpc.h"
 #include "ttime.h"
@@ -51,8 +51,9 @@ static void registerRequest(SRequestObj* pRequest) {
 
     int32_t total = atomic_add_fetch_32(&pSummary->totalRequests, 1);
     int32_t currentInst = atomic_add_fetch_32(&pSummary->currentRequests, 1);
-    tscDebug("0x%" PRIx64 " new Request from connObj:0x%" PRIx64 ", current:%d, app current:%d, total:%d, reqId:0x%"PRIx64, pRequest->self,
+    tscDebug("0x%" PRIx64 " new Request from connObj:0x%" PRIx64
-             pRequest->pTscObj->id, num, currentInst, total, pRequest->requestId);
+             ", current:%d, app current:%d, total:%d, reqId:0x%" PRIx64,
+             pRequest->self, pRequest->pTscObj->id, num, currentInst, total, pRequest->requestId);
   }
 }
 
@@ -66,11 +67,14 @@ static void deregisterRequest(SRequestObj* pRequest) {
   int32_t num = atomic_sub_fetch_32(&pTscObj->numOfReqs, 1);
 
   int64_t duration = taosGetTimestampMs() - pRequest->metric.start;
-  tscDebug("0x%"PRIx64" free Request from connObj: 0x%"PRIx64", reqId:0x%"PRIx64" elapsed:%"PRIu64" ms, current:%d, app current:%d", pRequest->self, pTscObj->id,
+  tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64 " elapsed:%" PRIu64
-      pRequest->requestId, duration, num, currentInst);
+           " ms, current:%d, app current:%d",
+           pRequest->self, pTscObj->id, pRequest->requestId, duration, num, currentInst);
   taosReleaseRef(clientConnRefPool, pTscObj->id);
 }
 
+
+
 // todo close the transporter properly
 void closeTransporter(STscObj *pTscObj) {
   if (pTscObj == NULL || pTscObj->pAppInfo->pTransporter == NULL) {
@@ -89,6 +93,7 @@ void* openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   rpcInit.label = "TSC";
   rpcInit.numOfThreads = numOfThread;
   rpcInit.cfp = processMsgFromServer;
+  rpcInit.pfp = persistConnForSpecificMsg;
   rpcInit.sessions = cfgGetItem(tscCfg, "maxConnections")->i32;
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.user = (char *)user;
@@ -269,7 +274,8 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
         cfg->cfgStatus = TAOS_CFG_CSTATUS_OPTION;
         tscInfo("set config file directory:%s", str);
       } else {
-        tscWarn("config option:%s, input value:%s, is configured by %s, use %s", cfg->option, str, tsCfgStatusStr[cfg->cfgStatus], (char *)cfg->ptr);
+        tscWarn("config option:%s, input value:%s, is configured by %s, use %s", cfg->option, str,
+                tsCfgStatusStr[cfg->cfgStatus], (char *)cfg->ptr);
       }
       break;
 
@@ -284,7 +290,8 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
         cfg->cfgStatus = TAOS_CFG_CSTATUS_OPTION;
         tscInfo("set shellActivityTimer:%d", tsShellActivityTimer);
       } else {
-        tscWarn("config option:%s, input value:%s, is configured by %s, use %d", cfg->option, str, tsCfgStatusStr[cfg->cfgStatus], *(int32_t *)cfg->ptr);
+        tscWarn("config option:%s, input value:%s, is configured by %s, use %d", cfg->option, str,
+                tsCfgStatusStr[cfg->cfgStatus], *(int32_t *)cfg->ptr);
       }
       break;
 
@@ -352,7 +359,8 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
           tscInfo("charset remains:%s", tsCharset);
         }
       } else {
-        tscWarn("config option:%s, input value:%s, is configured by %s, use %s", cfg->option, str, tsCfgStatusStr[cfg->cfgStatus], (char *)cfg->ptr);
+        tscWarn("config option:%s, input value:%s, is configured by %s, use %s", cfg->option, str,
+                tsCfgStatusStr[cfg->cfgStatus], (char *)cfg->ptr);
       }
       break;
     }
@@ -382,7 +390,8 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
           tscInfo("charset:%s not valid", str);
         }
       } else {
-        tscWarn("config option:%s, input value:%s, is configured by %s, use %s", cfg->option, str, tsCfgStatusStr[cfg->cfgStatus], (char *)cfg->ptr);
+        tscWarn("config option:%s, input value:%s, is configured by %s, use %s", cfg->option, str,
+                tsCfgStatusStr[cfg->cfgStatus], (char *)cfg->ptr);
       }
 
       break;
@@ -398,7 +407,8 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
         cfg->cfgStatus = TAOS_CFG_CSTATUS_OPTION;
         tscDebug("timezone set:%s, input:%s by taos_options", tsTimezone, str);
       } else {
-        tscWarn("config option:%s, input value:%s, is configured by %s, use %s", cfg->option, str, tsCfgStatusStr[cfg->cfgStatus], (char *)cfg->ptr);
+        tscWarn("config option:%s, input value:%s, is configured by %s, use %s", cfg->option, str,
+                tsCfgStatusStr[cfg->cfgStatus], (char *)cfg->ptr);
       }
       break;
 

source/client/src/clientImpl.c

@@ -363,7 +363,6 @@ static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) {
   pMsgSendInfo->fp = handleRequestRspFp[TMSG_INDEX(pMsgSendInfo->msgType)];
   pMsgSendInfo->param = pRequest;
 
-
   SConnectReq connectReq = {0};
   STscObj*    pObj = pRequest->pTscObj;
 
@@ -391,7 +390,9 @@ static void destroySendMsgInfo(SMsgSendInfo* pMsgBody) {
   tfree(pMsgBody->msgInfo.pData);
   tfree(pMsgBody);
 }
-
+bool persistConnForSpecificMsg(void* parenct, tmsg_t msgType) {
+  return msgType == TDMT_VND_QUERY_RSP || msgType == TDMT_VND_FETCH_RSP || msgType == TDMT_VND_RES_READY_RSP;
+}
 void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
   SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->ahandle;
   assert(pMsg->ahandle != NULL);

source/libs/index/src/index_fst.c

@@ -235,6 +235,7 @@ void fstStateCompileForOneTrans(FstCountingWriter* w, CompiledAddr addr, FstTran
   FstState st = fstStateCreate(OneTrans);
 
   fstStateSetCommInput(&st, trn->inp);
+
   bool    null = false;
   uint8_t inp = fstStateCommInput(&st, &null);
   if (null == true) {
@@ -936,6 +937,7 @@ FstLastTransition* fstLastTransitionCreate(uint8_t inp, Output out) {
 }
 
 void fstLastTransitionDestroy(FstLastTransition* trn) { free(trn); }
+
 void fstBuilderNodeUnfinishedLastCompiled(FstBuilderNodeUnfinished* unNode, CompiledAddr addr) {
   FstLastTransition* trn = unNode->last;
   if (trn == NULL) {

source/libs/index/src/index_fst_automation.c

@@ -16,24 +16,24 @@
 #include "index_fst_automation.h"
 
 StartWithStateValue* startWithStateValueCreate(StartWithStateKind kind, ValueType ty, void* val) {
-  StartWithStateValue* nsv = calloc(1, sizeof(StartWithStateValue));
+  StartWithStateValue* sv = calloc(1, sizeof(StartWithStateValue));
-  if (nsv == NULL) {
+  if (sv == NULL) {
     return NULL;
   }
 
-  nsv->kind = kind;
+  sv->kind = kind;
-  nsv->type = ty;
+  sv->type = ty;
   if (ty == FST_INT) {
-    nsv->val = *(int*)val;
+    sv->val = *(int*)val;
   } else if (ty == FST_CHAR) {
     size_t len = strlen((char*)val);
-    nsv->ptr = (char*)calloc(1, len + 1);
+    sv->ptr = (char*)calloc(1, len + 1);
-    memcpy(nsv->ptr, val, len);
+    memcpy(sv->ptr, val, len);
   } else if (ty == FST_ARRAY) {
     // TODO,
     // nsv->arr = taosArrayFromList()
   }
-  return nsv;
+  return sv;
 }
 void startWithStateValueDestroy(void* val) {
   StartWithStateValue* sv = (StartWithStateValue*)val;
@@ -146,11 +146,9 @@ AutomationCtx* automCtxCreate(void* data, AutomationType atype) {
   if (atype == AUTOMATION_ALWAYS) {
     int val = 0;
     sv = startWithStateValueCreate(Running, FST_INT, &val);
-    ctx->stdata = (void*)sv;
   } else if (atype == AUTOMATION_PREFIX) {
     int val = 0;
     sv = startWithStateValueCreate(Running, FST_INT, &val);
-    ctx->stdata = (void*)sv;
   } else if (atype == AUTMMATION_MATCH) {
   } else {
     // add more search type
@@ -160,9 +158,8 @@ AutomationCtx* automCtxCreate(void* data, AutomationType atype) {
   if (data != NULL) {
     char*  src = (char*)data;
     size_t len = strlen(src);
-    dst = (char*)malloc(len * sizeof(char) + 1);
+    dst = (char*)calloc(1, len * sizeof(char) + 1);
     memcpy(dst, src, len);
-    dst[len] = 0;
   }
 
   ctx->data = dst;

source/libs/index/test/fstUT.cc

@@ -99,6 +99,7 @@ class FstReadMemory {
     fstSliceDestroy(&skey);
     return ok;
   }
+
   bool GetWithTimeCostUs(const std::string& key, uint64_t* val, uint64_t* elapse) {
     int64_t s = taosGetTimestampUs();
     bool    ok = this->Get(key, val);
@@ -120,8 +121,6 @@ class FstReadMemory {
       printf("key: %s, val: %" PRIu64 "\n", key.c_str(), (uint64_t)(rt->out.out));
       swsResultDestroy(rt);
     }
-    for (size_t i = 0; i < result.size(); i++) {
-    }
     std::cout << std::endl;
     return true;
   }
@@ -137,7 +136,7 @@ class FstReadMemory {
     fstDestroy(_fst);
     fstSliceDestroy(&_s);
     writerCtxDestroy(_wc, false);
-    tfCleanup();
+    // tfCleanup();
   }
 
  private:
@@ -196,6 +195,10 @@ class TFst {
     }
     return fr->Get(k, v);
   }
+  bool Search(AutomationCtx* ctx, std::vector<uint64_t>& result) {
+    // add more
+    return fr->Search(ctx, result);
+  }
 
  private:
   FstWriter*     fw;
@@ -229,5 +232,9 @@ TEST_F(FstEnv, writeNormal) {
   assert(fst->Get("a", &val) == false);
   assert(fst->Get("aa", &val) == true);
   assert(val == 0);
+
+  std::vector<uint64_t> rlt;
+  AutomationCtx*        ctx = automCtxCreate((void*)"ab", AUTOMATION_ALWAYS);
+  assert(fst->Search(ctx, rlt) == true);
 }
-TEST_F(FstEnv, writeExcpet) {}
+TEST_F(FstEnv, WriteMillonrRecord) {}

source/libs/transport/inc/transportInt.h

@@ -66,6 +66,7 @@ typedef struct {
 
   void (*cfp)(void* parent, SRpcMsg*, SEpSet*);
   int (*afp)(void* parent, char* user, char* spi, char* encrypt, char* secret, char* ckey);
+  bool (*pfp)(void* parent, tmsg_t msgType);
 
   int32_t         refCount;
   void*           parent;

source/libs/transport/src/trans.c

@@ -29,7 +29,12 @@ void* rpcOpen(const SRpcInit* pInit) {
   if (pInit->label) {
     tstrncpy(pRpc->label, pInit->label, strlen(pInit->label) + 1);
   }
+
+  // register callback handle
   pRpc->cfp = pInit->cfp;
+  pRpc->afp = pInit->afp;
+  pRpc->pfp = pInit->pfp;
+
   if (pInit->connType == TAOS_CONN_SERVER) {
     pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
   } else {

source/libs/transport/src/transCli.c

@@ -134,8 +134,7 @@ static void clientHandleResp(SCliConn* conn) {
   rpcMsg.msgType = pHead->msgType;
   rpcMsg.ahandle = pCtx->ahandle;
 
-  if (rpcMsg.msgType == TDMT_VND_QUERY_RSP || rpcMsg.msgType == TDMT_VND_FETCH_RSP ||
+  if (pRpc->pfp != NULL && (pRpc->pfp)(pRpc->parent, rpcMsg.msgType)) {
-      rpcMsg.msgType == TDMT_VND_RES_READY_RSP) {
     rpcMsg.handle = conn;
     conn->persist = 1;
     tDebug("client conn %p persist by app", conn);
@@ -186,17 +185,12 @@ static void clientHandleExcept(SCliConn* pConn) {
     return;
   }
   SCliMsg*       pMsg = pConn->data;
-
-  tmsg_t msgType = TDMT_MND_CONNECT;
-  if (pMsg != NULL) {
-    msgType = pMsg->msg.msgType;
-  }
   STransConnCtx* pCtx = pMsg->ctx;
 
   SRpcMsg rpcMsg = {0};
   rpcMsg.ahandle = pCtx->ahandle;
   rpcMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
-  rpcMsg.msgType = msgType + 1;
+  rpcMsg.msgType = pMsg->msg.msgType + 1;
 
   if (pConn->push != NULL && pConn->ctnRdCnt != 0) {
     (*pConn->push->callback)(pConn->push->arg, &rpcMsg);
@@ -445,7 +439,7 @@ static void clientConnCb(uv_connect_t* req, int status) {
   addrlen = sizeof(pConn->locaddr);
   uv_tcp_getsockname((uv_tcp_t*)pConn->stream, (struct sockaddr*)&pConn->locaddr, &addrlen);
 
-  tTrace("client conn %p create", pConn);
+  tTrace("client conn %p connect to server successfully", pConn);
 
   assert(pConn->stream == req->handle);
   clientWrite(pConn);
@@ -524,6 +518,7 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
     struct sockaddr_in addr;
     uv_ip4_addr(pMsg->ctx->ip, pMsg->ctx->port, &addr);
     // handle error in callback if fail to connect
+    tTrace("client conn %p try to connect to %s:%d", conn, pMsg->ctx->ip, pMsg->ctx->port);
     uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, clientConnCb);
   }
 

source/libs/transport/src/transSrv.c

@@ -413,11 +413,6 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
     } else {
       uvStartSendResp(msg);
     }
-    // uv_buf_t wb;
-    // uvPrepareSendData(msg, &wb);
-    // uv_timer_stop(conn->pTimer);
-
-    // uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb);
   }
 }
 static void uvAcceptAsyncCb(uv_async_t* async) {
@@ -490,7 +485,6 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
   pConn->pTimer->data = pConn;
 
   pConn->hostThrd = pThrd;
-  // pConn->pWorkerAsync = pThrd->workerAsync;  // thread safty
 
   // init client handle
   pConn->pTcp = (uv_tcp_t*)malloc(sizeof(uv_tcp_t));
@@ -730,14 +724,9 @@ void destroyWorkThrd(SWorkThrdObj* pThrd) {
 }
 void sendQuitToWorkThrd(SWorkThrdObj* pThrd) {
   SSrvMsg* srvMsg = calloc(1, sizeof(SSrvMsg));
-
-  // pthread_mutex_lock(&pThrd->msgMtx);
-  // QUEUE_PUSH(&pThrd->msg, &srvMsg->q);
-  // pthread_mutex_unlock(&pThrd->msgMtx);
   tDebug("send quit msg to work thread");
 
   transSendAsync(pThrd->asyncPool, &srvMsg->q);
-  // uv_async_send(pThrd->workerAsync);
 }
 
 void taosCloseServer(void* arg) {
@@ -774,19 +763,12 @@ void rpcSendResponse(const SRpcMsg* pMsg) {
   SSrvMsg* srvMsg = calloc(1, sizeof(SSrvMsg));
   srvMsg->pConn = pConn;
   srvMsg->msg = *pMsg;
-
-  // pthread_mutex_lock(&pThrd->msgMtx);
-  // QUEUE_PUSH(&pThrd->msg, &srvMsg->q);
-  // pthread_mutex_unlock(&pThrd->msgMtx);
-
   tTrace("server conn %p start to send resp", pConn);
   transSendAsync(pThrd->asyncPool, &srvMsg->q);
-  // uv_async_send(pThrd->workerAsync);
 }
 
 int rpcGetConnInfo(void* thandle, SRpcConnInfo* pInfo) {
   SSrvConn* pConn = thandle;
-  // struct sockaddr* pPeerName = &pConn->peername;
 
   struct sockaddr_in addr = pConn->addr;
   pInfo->clientIp = (uint32_t)(addr.sin_addr.s_addr);