Feat/ts 5436 3.0 acore rpc

cmake/cmake.platform

@@ -218,5 +218,9 @@ else ()
 endif()
 MESSAGE(STATUS "DEPS_DIR: " ${TD_DEPS_DIR})
 
+if(${TD_ACORE})
+  add_definitions(-DTD_ACORE)
+endif(${TD_ACORE})
+
 MESSAGE("C Compiler: ${CMAKE_C_COMPILER} (${CMAKE_C_COMPILER_ID}, ${CMAKE_C_COMPILER_VERSION})")
 MESSAGE("CXX Compiler: ${CMAKE_CXX_COMPILER} (${CMAKE_C_COMPILER_ID}, ${CMAKE_CXX_COMPILER_VERSION})")

include/common/tglobal.h

@@ -124,6 +124,7 @@ extern bool    tsEnableWhiteList;
 extern int64_t tsDndStart;
 extern int64_t tsDndStartOsUptime;
 extern int64_t tsDndUpTime;
+extern int64_t tsDndStarted;
 
 // dnode misc
 extern uint32_t tsEncryptionKeyChksum;
@@ -243,6 +244,9 @@ extern int64_t tsmaDataDeleteMark;
 // wal
 extern int64_t tsWalFsyncDataSizeLimit;
 
+// misc
+extern bool tsAcoreOS;
+
 // internal
 extern int32_t tsTransPullupInterval;
 extern int32_t tsCompactPullupInterval;
@@ -272,9 +276,9 @@ extern bool tsExperimental;
 // #define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize)
 
 int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd,
-                      const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc);
-int32_t taosReadDataFolder(const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl,
-                           SArray *pArgs);
+                      const char *envFile, char *apolloUrl, SArray *pArgs, ELogMode mode);
+int32_t taosReadDataFolder(const char *cfgDir, const char **envCmd,
+                      const char *envFile, char *apolloUrl, SArray *pArgs);
 int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl, SArray *pArgs,
                     bool tsc);
 void    taosCleanupCfg();

include/libs/transport/trpc.h

@@ -19,6 +19,7 @@
 extern "C" {
 #endif
 
+#ifndef TD_ACORE
 #include <stdbool.h>
 #include <stdint.h>
 #include "taosdef.h"
@@ -190,6 +191,193 @@ int32_t rpcUtilSIpRangeToStr(SIpV4Range *pRange, char *buf);
 int32_t rpcUtilSWhiteListToStr(SIpWhiteList *pWhiteList, char **ppBuf);
 int32_t rpcCvtErrCode(int32_t code);
 
+#else
+#include <stdbool.h>
+#include <stdint.h>
+#include "taosdef.h"
+#include "tmsg.h"
+#include "ttrace.h"
+
+#define TAOS_CONN_SERVER 0
+#define TAOS_CONN_CLIENT 1
+#define IsReq(pMsg)      (pMsg->msgType & 1U)
+
+extern int32_t tsRpcHeadSize;
+
+typedef struct {
+  uint32_t clientIp;
+  uint16_t clientPort;
+  int64_t  applyIndex;
+  uint64_t applyTerm;
+  char     user[TSDB_USER_LEN];
+} SRpcConnInfo;
+
+typedef enum {
+  TD_ACORE_CLIENT = 1,
+  TD_ACORE_DSVR_CLIENT = 2,
+  TD_ACORE_DSVR_STA_CLIENT = 4,
+  TD_ACORE_DSVR_SYNC_CLIENT = 8,
+  TD_ACORE_DSVR = 16
+} RPC_TYPE;
+
+typedef struct SRpcHandleInfo {
+  // rpc info
+  void   *handle;         // rpc handle returned to app
+  int64_t refId;          // refid, used by server
+  int8_t  noResp;         // has response or not(default 0, 0: resp, 1: no resp)
+  int8_t  persistHandle;  // persist handle or not
+  int8_t  hasEpSet;
+  int32_t cliVer;
+
+  // app info
+  void *ahandle;  // app handle set by client
+  void *wrapper;  // wrapper handle
+  void *node;     // node mgmt handle
+
+  // resp info
+  void   *rsp;
+  int32_t rspLen;
+
+  STraceId traceId;
+
+  SRpcConnInfo conn;
+  int8_t       forbiddenIp;
+  int8_t       notFreeAhandle;
+  int8_t       compressed;
+  int16_t      connType;
+  int64_t      seq;
+  int64_t      qId;
+  int32_t      msgType;
+  void        *reqWithSem;
+  int32_t      refIdMgt;
+} SRpcHandleInfo;
+
+typedef struct SRpcMsg {
+  tmsg_t         msgType;
+  void          *pCont;
+  int32_t        contLen;
+  int32_t        code;
+  int32_t        type;
+  void          *parent;
+  SRpcHandleInfo info;
+
+} SRpcMsg;
+
+typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *epset);
+typedef bool (*RpcRfp)(int32_t code, tmsg_t msgType);
+typedef bool (*RpcTfp)(int32_t code, tmsg_t msgType);
+typedef bool (*RpcFFfp)(tmsg_t msgType);
+typedef bool (*RpcNoDelayfp)(tmsg_t msgType);
+typedef void (*RpcDfp)(void *ahandle);
+
+typedef struct SRpcInit {
+  char     localFqdn[TSDB_FQDN_LEN];
+  uint16_t localPort;     // local port
+  char    *label;         // for debug purpose
+  int32_t  numOfThreads;  // number of threads to handle connections
+  int32_t  sessions;      // number of sessions allowed
+  int8_t   connType;      // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
+  int32_t  idleTime;      // milliseconds, 0 means idle timer is disabled
+  int32_t  compatibilityVer;
+
+  int32_t retryMinInterval;  // retry init interval
+  int32_t retryStepFactor;   // retry interval factor
+  int32_t retryMaxInterval;  // retry max interval
+  int64_t retryMaxTimeout;
+
+  int32_t failFastThreshold;
+  int32_t failFastInterval;
+
+  int32_t compressSize;  // -1: no compress, 0 : all data compressed, size: compress data if larger than size
+  int8_t  encryption;    // encrypt or not
+
+  // the following is for client app ecurity only
+  char *user;  // user name
+
+  // call back to process incoming msg
+  RpcCfp cfp;
+
+  // retry not not for particular msg
+  RpcRfp rfp;
+
+  // set up timeout for particular msg
+  RpcTfp tfp;
+
+  // destroy client ahandle;
+  RpcDfp dfp;
+  // fail fast fp
+  RpcFFfp ffp;
+
+  RpcNoDelayfp noDelayFp;
+
+  int32_t connLimitNum;
+  int32_t connLimitLock;
+  int32_t timeToGetConn;
+  int8_t  supportBatch;  // 0: no batch, 1. batch
+  int32_t batchSize;
+  int8_t  notWaitAvaliableConn;  // 1: wait to get, 0: no wait
+  int32_t shareConnLimit;
+  int8_t  shareConn;  // 0: no share, 1. share
+  int8_t  startReadTimer;
+  int64_t readTimeout;  // s
+  void   *parent;
+
+} SRpcInit;
+
+typedef struct {
+  void *val;
+  int32_t (*clone)(void *src, void **dst);
+} SRpcCtxVal;
+
+typedef struct {
+  int32_t msgType;
+  void   *val;
+  int32_t (*clone)(void *src, void **dst);
+} SRpcBrokenlinkVal;
+
+typedef struct {
+  SHashObj         *args;
+  SRpcBrokenlinkVal brokenVal;
+  void (*freeFunc)(const void *arg);
+  int64_t st;
+} SRpcCtx;
+
+int32_t rpcInit();
+void    rpcCleanup();
+
+void *rpcOpen(const SRpcInit *pRpc);
+void  rpcClose(void *);
+void  rpcCloseImpl(void *);
+void *rpcMallocCont(int64_t contLen);
+void  rpcFreeCont(void *pCont);
+void *rpcReallocCont(void *ptr, int64_t contLen);
+
+// Because taosd supports multi-process mode
+// These functions should not be used on the server side
+// Please use tmsg<xx> functions, which are defined in tmsgcb.h
+int32_t rpcSendRequest(void *thandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t *rid);
+int32_t rpcSendResponse(SRpcMsg *pMsg);
+int32_t rpcRegisterBrokenLinkArg(SRpcMsg *msg);
+int32_t rpcReleaseHandle(void *handle, int8_t type);  // just release conn to rpc instance, no close sock
+
+// These functions will not be called in the child process
+int32_t rpcSendRequestWithCtx(void *thandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t *rid, SRpcCtx *ctx);
+int32_t rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp);
+int32_t rpcSendRecvWithTimeout(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp, int8_t *epUpdated,
+                               int32_t timeoutMs);
+
+int32_t rpcFreeConnById(void *shandle, int64_t connId);
+
+int32_t rpcSetDefaultAddr(void *thandle, const char *ip, const char *fqdn);
+int32_t rpcAllocHandle(int64_t *refId);
+int32_t rpcSetIpWhite(void *thandl, void *arg);
+
+int32_t rpcUtilSIpRangeToStr(SIpV4Range *pRange, char *buf);
+
+int32_t rpcUtilSWhiteListToStr(SIpWhiteList *pWhiteList, char **ppBuf);
+int32_t rpcCvtErrCode(int32_t code);
+
+#endif
 #ifdef __cplusplus
 }
 #endif

include/os/osSocket.h

@@ -18,17 +18,17 @@
 
 // If the error is in a third-party library, place this header file under the third-party library header file.
 // When you want to use this feature, you should find or add the same function in the following section.
-#ifndef ALLOW_FORBID_FUNC
-#define socket       SOCKET_FUNC_TAOS_FORBID
-#define bind         BIND_FUNC_TAOS_FORBID
-#define listen       LISTEN_FUNC_TAOS_FORBID
-#define accept       ACCEPT_FUNC_TAOS_FORBID
-#define epoll_create EPOLL_CREATE_FUNC_TAOS_FORBID
-#define epoll_ctl    EPOLL_CTL_FUNC_TAOS_FORBID
-#define epoll_wait   EPOLL_WAIT_FUNC_TAOS_FORBID
-#define inet_addr    INET_ADDR_FUNC_TAOS_FORBID
-#define inet_ntoa    INET_NTOA_FUNC_TAOS_FORBID
-#endif
+// #ifndef ALLOW_FORBID_FUNC
+// #define socket       SOCKET_FUNC_TAOS_FORBID
+// #define bind         BIND_FUNC_TAOS_FORBID
+// #define listen       LISTEN_FUNC_TAOS_FORBID
+// #define accept       ACCEPT_FUNC_TAOS_FORBID
+// #define epoll_create EPOLL_CREATE_FUNC_TAOS_FORBID
+// #define epoll_ctl    EPOLL_CTL_FUNC_TAOS_FORBID
+// #define epoll_wait   EPOLL_WAIT_FUNC_TAOS_FORBID
+// #define inet_addr    INET_ADDR_FUNC_TAOS_FORBID
+// #define inet_ntoa    INET_NTOA_FUNC_TAOS_FORBID
+// #endif
 
 #if defined(WINDOWS)
 #if BYTE_ORDER == LITTLE_ENDIAN

include/util/tlog.h

@@ -34,6 +34,12 @@ typedef enum {
   DEBUG_FILE = 128
 } ELogLevel;
 
+typedef enum {
+  LOG_MODE_TAOSC = 1,
+  LOG_MODE_TAOSD = 2,
+  LOG_MODE_BOTH = 3,  // LOG_MODE_TAOSC | LOG_MODE_TAOSD
+} ELogMode;
+
 typedef void (*LogFp)(int64_t ts, ELogLevel level, const char *content);
 
 extern bool    tsLogEmbedded;
@@ -73,7 +79,7 @@ extern int32_t simDebugFlag;
 
 extern int32_t tqClientDebugFlag;
 int32_t taosInitLogOutput(const char **ppLogName);
-int32_t taosInitLog(const char *logName, int32_t maxFiles, bool tsc);
+int32_t taosInitLog(const char *logName, int32_t maxFiles, ELogMode mode);
 void    taosCloseLog();
 void    taosResetLog();
 void    taosDumpData(uint8_t *msg, int32_t len);

source/client/src/clientEnv.c

@@ -960,13 +960,13 @@ void taos_init_imp(void) {
 
   const char *logName = CUS_PROMPT "slog";
   ENV_ERR_RET(taosInitLogOutput(&logName), "failed to init log output");
-  if (taosCreateLog(logName, 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
+  if (taosCreateLog(logName, 10, configDir, NULL, NULL, NULL, NULL, tsAcoreOS ? LOG_MODE_BOTH : LOG_MODE_TAOSC) != 0) {
     (void)printf(" WARING: Create %s failed:%s. configDir=%s\n", logName, strerror(errno), configDir);
     tscInitRes = -1;
     return;
   }
 
-  ENV_ERR_RET(taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1), "failed to init cfg");
+  ENV_ERR_RET(taosInitCfg(configDir, NULL, NULL, NULL, NULL, tsAcoreOS ? 0 : 1), "failed to init cfg");
 
   initQueryModuleMsgHandle();
   ENV_ERR_RET(taosConvInit(), "failed to init conv");

source/client/src/clientMain.c

@@ -93,7 +93,7 @@ void taos_cleanup(void) {
 
   tscInfo("all local resources released");
   taosCleanupCfg();
-  taosCloseLog();
+  if (!tsAcoreOS) taosCloseLog();
 }
 
 static setConfRet taos_set_config_imp(const char *config) {

source/common/src/tglobal.c

@@ -100,6 +100,7 @@ int32_t tsArbSetAssignedTimeoutSec = 30;
 int64_t tsDndStart = 0;
 int64_t tsDndStartOsUptime = 0;
 int64_t tsDndUpTime = 0;
+int64_t tsDndStarted = 0;
 
 // dnode misc
 uint32_t tsEncryptionKeyChksum = 0;
@@ -278,6 +279,9 @@ int32_t tsTtlFlushThreshold = 100;   /* maximum number of dirty items in memory.
                                       */
 int32_t tsTtlBatchDropNum = 10000;   // number of tables dropped per batch
 
+// misc
+bool tsAcoreOS = false;
+
 // internal
 int32_t tsTransPullupInterval = 2;
 int32_t tsCompactPullupInterval = 10;
@@ -1731,24 +1735,26 @@ int32_t taosSetReleaseCfg(SConfig *pCfg);
 
 static int32_t taosSetAllDebugFlag(SConfig *pCfg, int32_t flag);
 
+static int8_t tsLogCreated = 0;
+
 int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd,
-                      const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc) {
+                      const char *envFile, char *apolloUrl, SArray *pArgs, ELogMode mode) {
   int32_t  code = TSDB_CODE_SUCCESS;
   int32_t  lino = 0;
   SConfig *pCfg = NULL;
 
+  if (atomic_val_compare_exchange_8(&tsLogCreated, 0, 1) != 0) return 0;
+
   if (tsCfg == NULL) {
     TAOS_CHECK_GOTO(osDefaultInit(), &lino, _exit);
   }
 
   TAOS_CHECK_GOTO(cfgInit(&pCfg), &lino, _exit);
 
-  if (tsc) {
-    tsLogEmbedded = 0;
-    TAOS_CHECK_GOTO(taosAddClientLogCfg(pCfg), &lino, _exit);
-  } else {
-    tsLogEmbedded = 1;
-    TAOS_CHECK_GOTO(taosAddClientLogCfg(pCfg), &lino, _exit);
+  tsLogEmbedded = (mode & LOG_MODE_TAOSC) ? 0 : 1;
+  TAOS_CHECK_GOTO(taosAddClientLogCfg(pCfg), &lino, _exit);
+
+  if (mode & LOG_MODE_TAOSD) {
     TAOS_CHECK_GOTO(taosAddServerLogCfg(pCfg), &lino, _exit);
   }
 
@@ -1762,10 +1768,8 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
     goto _exit;
   }
 
-  if (tsc) {
-    TAOS_CHECK_GOTO(taosSetClientLogCfg(pCfg), &lino, _exit);
-  } else {
-    TAOS_CHECK_GOTO(taosSetClientLogCfg(pCfg), &lino, _exit);
+  TAOS_CHECK_GOTO(taosSetClientLogCfg(pCfg), &lino, _exit);
+  if (mode & LOG_MODE_TAOSD) {
     TAOS_CHECK_GOTO(taosSetServerLogCfg(pCfg), &lino, _exit);
   }
 
@@ -1778,7 +1782,7 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
     goto _exit;
   }
 
-  if ((code = taosInitLog(logname, logFileNum, tsc)) != 0) {
+  if ((code = taosInitLog(logname, logFileNum, mode)) != 0) {
     (void)printf("failed to init log file since %s\n", tstrerror(code));
     goto _exit;
   }