feat: determine whether the dispatcher is threadlocal by taos.cfg

packaging/cfg/taos.cfg

@@ -311,11 +311,14 @@ keepColumnName            1
 # unit Hour. Latency of data migration
 # keepTimeOffset     0
 
-# taosc async batching insertion
+# enable taosc async batching insertion
 # asyncBatchEnable   1
 
-# taosc async insertion batchsize
+# enable thread local batch dispatcher
+# asyncBatchThreadLocal   0
+
+# taosc async insertion batch size, maximum 65535
 # asyncBatchSize     256
 
-# taosc async batching timeout in milliseconds
+# taosc async batching timeout in milliseconds, maximum 2048
 # asyncBatchTimeout  5
\ No newline at end of file

src/client/inc/tscBulkWrite.h

@@ -21,7 +21,6 @@ extern "C" {
 #endif
 
 #include "tarray.h"
-#include "tlist.h"
 #include "tthread.h"
 
 /**
@@ -29,11 +28,14 @@ extern "C" {
  * and merge them into single statement.
  */
 typedef struct SAsyncBulkWriteDispatcher {
-  // the buffer to store the insertion statements. equivalent to SList<SSqlObj*>.
-  SList* buffer;
+  // the buffer to store the insertion statements. equivalent to SArray<SSqlObj*>.
+  SArray* buffer;
 
   // the mutex to protect the buffer.
   pthread_mutex_t mutex;
+  
+  // the cond to signal to background thread.
+  pthread_cond_t cond;
 
   // the background thread to manage batching timeout.
   pthread_t background;
@@ -131,9 +133,10 @@ bool tscSupportBulkInsertion(SSqlObj* pSql);
 bool dispatcherTryBatching(SAsyncBulkWriteDispatcher* dispatcher, SSqlObj* pSql);
 
 /**
- * A thread local version of SAsyncBulkWriteDispatcher.
+ * A holder of SAsyncBulkWriteDispatcher. Call dispatcherAcquire(...) to get the SAsyncBulkWriteDispatcher
+ * instance. This holder will manage the life cycle of SAsyncBulkWriteDispatcher.
  */
-typedef struct SThreadLocalDispatcher {
+typedef struct SDispatcherHolder {
   pthread_key_t key;
 
   // the maximum number of insertion rows in a batch.
@@ -141,32 +144,40 @@ typedef struct SThreadLocalDispatcher {
 
   // the batching timeout in milliseconds.
   int32_t timeoutMs;
+  
+  // specifies whether the dispatcher is thread local, if the dispatcher is not
+  // thread local, we will use the global dispatcher below.
+  bool isThreadLocal;
+  
+  // the global dispatcher, if thread local enabled, global will be set to NULL.
+  SAsyncBulkWriteDispatcher * global;
 
-} SThreadLocalDispatcher;
+} SDispatcherHolder;
 
 /**
- * Create a thread local SAsyncBulkWriteDispatcher variable.
+ * Create a holder of SAsyncBulkWriteDispatcher.
  *
  * @param batchSize     the batchSize of SAsyncBulkWriteDispatcher.
  * @param timeoutMs     the timeoutMs of SAsyncBulkWriteDispatcher.
- * @return the thread local SAsyncBulkWriteDispatcher.
+ * @param isThreadLocal specifies whether the dispatcher is thread local.
+ * @return the SAsyncBulkWriteDispatcher holder.
  */
-SThreadLocalDispatcher* createThreadLocalDispatcher(int32_t batchSize, int32_t timeoutMs);
+SDispatcherHolder* createDispatcherHolder(int32_t batchSize, int32_t timeoutMs, bool isThreadLocal);
 
 /**
- * Destroy the thread local SAsyncBulkWriteDispatcher variable.
+ * Destroy the holder of SAsyncBulkWriteDispatcher.
  * (will destroy all the instances of SAsyncBulkWriteDispatcher in the thread local variable)
  *
- * @param dispatcher  the thread local SAsyncBulkWriteDispatcher variable.
+ * @param holder  the holder of SAsyncBulkWriteDispatcher.
  */
-void destroyThreadLocalDispatcher(SThreadLocalDispatcher* dispatcher);
+void destroyDispatcherHolder(SDispatcherHolder* holder);
 
 /**
- * Get the thread local instance of SAsyncBulkWriteDispatcher.
- * @param dispatcher  the thread local SAsyncBulkWriteDispatcher variable.
- * @return the thread local SAsyncBulkWriteDispatcher.
+ * Get an instance of SAsyncBulkWriteDispatcher.
+ * @param holder  the holder of SAsyncBulkWriteDispatcher.
+ * @return the SAsyncBulkWriteDispatcher instance.
  */
-SAsyncBulkWriteDispatcher* dispatcherThreadLocal(SThreadLocalDispatcher* dispatcher);
+SAsyncBulkWriteDispatcher* dispatcherAcquire(SDispatcherHolder* holder);
 
 #ifdef __cplusplus
 }

src/client/inc/tsclient.h

@@ -509,8 +509,9 @@ void waitForQueryRsp(void *param, TAOS_RES *tres, int code);
  * 
  * @param batchSize  the batchSize of async bulk write dispatcher.
  * @param timeoutMs  the timeout of batching in milliseconds.
+ * @param isThreadLocal specifies whether the dispatcher is thread local.
  */
-void tscInitAsyncDispatcher(int32_t batchSize, int32_t timeoutMs);
+void tscInitAsyncDispatcher(int32_t batchSize, int32_t timeoutMs, bool isThreadLocal);
 
 /**
  * Destroy the async auto batch dispatcher.
@@ -546,8 +547,8 @@ extern SHashObj  *tscTableMetaMap;
 extern SCacheObj *tscVgroupListBuf;
 
 // forward declaration.
-typedef struct SThreadLocalDispatcher    SThreadLocalDispatcher;
-extern SThreadLocalDispatcher           *tscDispatcher;
+typedef struct SDispatcherHolder    SDispatcherHolder;
+extern SDispatcherHolder           *tscDispatcher;
 extern int   tscObjRef;
 extern void *tscTmr;
 extern void *tscQhandle;

src/client/src/tscAsync.c

@@ -398,7 +398,7 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, __async_cb_func_t fp, void* para
   }
   
   if (tscDispatcher != NULL) {
-    SAsyncBulkWriteDispatcher* dispatcher = dispatcherThreadLocal(tscDispatcher);
+    SAsyncBulkWriteDispatcher* dispatcher = dispatcherAcquire(tscDispatcher);
     if (dispatcherTryBatching(dispatcher, pSql)) {
       taosReleaseRef(tscObjRef, pSql->self);
       tscDebug("sql obj %p has been buffer in insert buffer", pSql);

src/client/src/tscBulkWrite.c

 /*
-* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
-*
-* This program is free software: you can use, redistribute, and/or modify
-* it under the terms of the GNU Affero General Public License, version 3
-* or later ("AGPL"), as published by the Free Software Foundation.
-*
-* This program is distributed in the hope that it will be useful, but WITHOUT
-* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
-* FITNESS FOR A PARTICULAR PURPOSE.
-*
-* You should have received a copy of the GNU Affero General Public License
-* along with this program. If not, see <http://www.gnu.org/licenses/>.
-*/
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
 
 #include "osAtomic.h"
 
-#include "tsclient.h"
 #include "tscBulkWrite.h"
-#include "tscSubquery.h"
 #include "tscLog.h"
+#include "tscSubquery.h"
+#include "tsclient.h"
 
 /**
  * Represents the callback function and its context.
@@ -50,7 +50,7 @@ inline static int32_t statementGetInsertionRows(SSqlObj* pSql) { return pSql->cm
  * @param pSql  the sql object.
  * @param code  the error code of the error result.
  */
-inline static void tscReturnsError(SSqlObj *pSql, int code) {
+inline static void tscReturnsError(SSqlObj* pSql, int code) {
   if (pSql == NULL) {
     return;
   }
@@ -122,11 +122,9 @@ int32_t dispatcherStatementMerge(SArray* statements, SSqlObj** result) {
   // initialize the callback context.
   context->count = count;
   for (size_t i = 0; i < count; ++i) {
-    SSqlObj*  statement = *((SSqlObj**)taosArrayGet(statements, i));
-    Runnable* callback = &context->runnable[i];
-
-    callback->fp = statement->fp;
-    callback->param = statement->param;
+    SSqlObj* statement = *((SSqlObj**)taosArrayGet(statements, i));
+    context->runnable[i].fp = statement->fp;
+    context->runnable[i].param = statement->param;
   }
 
   // merge the statements into single one.
@@ -149,47 +147,39 @@ SArray* dispatcherPollAll(SAsyncBulkWriteDispatcher* dispatcher) {
   if (!atomic_load_32(&dispatcher->bufferSize)) {
     return NULL;
   }
-  
+
   pthread_mutex_lock(&dispatcher->mutex);
-  SArray* statements = taosArrayInit(atomic_load_32(&dispatcher->bufferSize), sizeof(SSqlObj*));
+  SArray* statements = taosArrayDup(dispatcher->buffer);
   if (statements == NULL) {
     pthread_mutex_unlock(&dispatcher->mutex);
     tscError("failed to poll all items: out of memory");
     return NULL;
   }
 
-  // get all the sql statements from the buffer.
-  while (true) {
-    SListNode* node = tdListPopHead(dispatcher->buffer);
-    if (!node) {
-      break;
-    }
-
-    // get the SSqlObj* from the node.
-    SSqlObj* item = *((SSqlObj **) node->data);
-    listNodeFree(node);
-    atomic_fetch_sub_32(&dispatcher->bufferSize, 1);
-    atomic_fetch_sub_32(&dispatcher->currentSize, statementGetInsertionRows(item));
-    taosArrayPush(statements, &item);
-  }
-  
+  atomic_store_32(&dispatcher->bufferSize, 0);
+  atomic_store_32(&dispatcher->currentSize, 0);
+  taosArrayClear(dispatcher->buffer);
   pthread_mutex_unlock(&dispatcher->mutex);
+
   return statements;
 }
 
 int32_t dispatcherTryOffer(SAsyncBulkWriteDispatcher* dispatcher, SSqlObj* pSql) {
-  // the buffer is full.
+  // pre-check: the buffer is full.
   if (atomic_load_32(&dispatcher->currentSize) >= dispatcher->batchSize) {
     return -1;
   }
 
-  // offer the node to the buffer.
   pthread_mutex_lock(&dispatcher->mutex);
-  if (tdListAppend(dispatcher->buffer, &pSql)) {
+
+  // double-check: the buffer is full.
+  if (atomic_load_32(&dispatcher->currentSize) >= dispatcher->batchSize) {
     pthread_mutex_unlock(&dispatcher->mutex);
     return -1;
   }
 
+  taosArrayPush(dispatcher->buffer, pSql);
+
   tscDebug("sql obj %p has been write to insert buffer", pSql);
 
   atomic_fetch_add_32(&dispatcher->bufferSize, 1);
@@ -218,7 +208,7 @@ void dispatcherExecute(SArray* statements) {
   taosArrayDestroy(&statements);
   return;
 
-  _error:
+_error:
   tscError("send async batch sql obj failed, reason: %s", tstrerror(code));
 
   // handling the failures.
@@ -229,6 +219,32 @@ void dispatcherExecute(SArray* statements) {
   taosArrayDestroy(&statements);
 }
 
+/**
+ * Get the timespec after `millis` ms
+ *
+ * @param t         the timespec.
+ * @param millis    the duration in milliseconds.
+ * @return          the timespec after `millis` ms.
+ */
+static inline struct timespec afterMillis(struct timespec t, int32_t millis) {
+  t.tv_nsec += millis * 1000000L;
+  t.tv_sec += t.tv_nsec / 1000000000L;
+  t.tv_nsec %= 1000000000L;
+  return t;
+}
+
+/**
+ * Get the duration in milliseconds from timespec s to timespec t.
+ * @param s the start timespec.
+ * @param t the end timespec.
+ * @return  the duration in milliseconds.
+ */
+static inline int64_t durationMillis(struct timespec s, struct timespec t) {
+  int64_t d = (t.tv_sec - s.tv_sec) * 1000;
+  d += (t.tv_nsec - s.tv_nsec) / 1000000L;
+  return d;
+}
+
 /**
  * The thread to manage batching timeout.
  */
@@ -237,21 +253,34 @@ static void* dispatcherTimeoutCallback(void* arg) {
   setThreadName("tscBackground");
 
   while (!atomic_load_8(&dispatcher->shutdown)) {
-    int64_t t0 = taosGetTimestampNs();
+    struct timespec t1, t2;
+    clock_gettime(CLOCK_REALTIME, &t1);
 
     atomic_store_8(&dispatcher->exclusive, true);
     SArray* statements = dispatcherPollAll(dispatcher);
     atomic_store_8(&dispatcher->exclusive, false);
 
     dispatcherExecute(statements);
-
-    int64_t t1 = taosGetTimestampNs();
-    int64_t durationMs = (t1 - t0) / 1000000;
+    clock_gettime(CLOCK_REALTIME, &t2);
 
     // Similar to scheduleAtFixedRate in Java, if the execution time exceed
     // `timeoutMs` milliseconds, then there will be no sleep.
-    if (durationMs < dispatcher->timeoutMs) {
-      taosMsleep((int32_t)(dispatcher->timeoutMs - durationMs));
+    struct timespec t3 = afterMillis(t1, dispatcher->timeoutMs);
+    if (durationMillis(t2, t3) > 0) {
+      if (pthread_mutex_timedlock(&dispatcher->mutex, &t3)) {
+        continue;
+      }
+
+      while (true) {
+        if (atomic_load_8(&dispatcher->shutdown)) {
+          break;
+        }
+
+        if (pthread_cond_timedwait(&dispatcher->cond, &dispatcher->mutex, &t3)) {
+          break;
+        }
+      }
+      pthread_mutex_unlock(&dispatcher->mutex);
     }
   }
   return NULL;
@@ -272,18 +301,19 @@ SAsyncBulkWriteDispatcher* createAsyncBulkWriteDispatcher(int32_t batchSize, int
   atomic_store_8(&dispatcher->exclusive, false);
 
   // init the buffer.
-  dispatcher->buffer = tdListNew(sizeof(SSqlObj*));
+  dispatcher->buffer = taosArrayInit(batchSize, sizeof(SSqlObj*));
   if (!dispatcher->buffer) {
     tfree(dispatcher);
     return NULL;
   }
 
-  // init the mutex.
+  // init the mutex and the cond.
   pthread_mutex_init(&dispatcher->mutex, NULL);
+  pthread_cond_init(&dispatcher->cond, NULL);
 
   // init background thread.
   if (pthread_create(&dispatcher->background, NULL, dispatcherTimeoutCallback, dispatcher)) {
-    tdListFree(dispatcher->buffer);
+    taosArrayDestroy(&dispatcher->buffer);
     tfree(dispatcher);
     return NULL;
   }
@@ -295,21 +325,24 @@ void destroyAsyncDispatcher(SAsyncBulkWriteDispatcher* dispatcher) {
   if (dispatcher == NULL) {
     return;
   }
-  
-  // mark shutdown.
+
+  // mark shutdown, signal shutdown to timeout thread.
+  pthread_mutex_lock(&dispatcher->mutex);
   atomic_store_8(&dispatcher->shutdown, true);
-  
+  pthread_cond_signal(&dispatcher->cond);
+  pthread_mutex_unlock(&dispatcher->mutex);
+
   // make sure the timeout thread exit.
   pthread_join(dispatcher->background, NULL);
-  
+
   // poll and send all the statements in the buffer.
   while (atomic_load_32(&dispatcher->bufferSize)) {
     SArray* statements = dispatcherPollAll(dispatcher);
     dispatcherExecute(statements);
   }
-  
+
   // destroy the buffer.
-  tdListFree(dispatcher->buffer);
+  taosArrayDestroy(&dispatcher->buffer);
 
   // destroy the mutex.
   pthread_mutex_destroy(&dispatcher->mutex);
@@ -374,52 +407,69 @@ bool dispatcherTryBatching(SAsyncBulkWriteDispatcher* dispatcher, SSqlObj* pSql)
 }
 
 /**
- * Destroy the SAsyncBulkWriteDispatcher create by SThreadLocalDispatcher.
- * @param arg 
+ * Destroy the SAsyncBulkWriteDispatcher create by SDispatcherHolder.
+ * @param arg the thread local SAsyncBulkWriteDispatcher.
  */
 static void destroyDispatcher(void* arg) {
   SAsyncBulkWriteDispatcher* dispatcher = arg;
   if (!dispatcher) {
     return;
   }
-  
+
   destroyAsyncDispatcher(dispatcher);
 }
 
-SThreadLocalDispatcher* createThreadLocalDispatcher(int32_t batchSize, int32_t timeoutMs) {
-  SThreadLocalDispatcher* dispatcher = calloc(1, sizeof(SThreadLocalDispatcher));
+SDispatcherHolder* createDispatcherHolder(int32_t batchSize, int32_t timeoutMs, bool isThreadLocal) {
+  SDispatcherHolder* dispatcher = calloc(1, sizeof(SDispatcherHolder));
   if (!dispatcher) {
     return NULL;
   }
 
   dispatcher->batchSize = batchSize;
   dispatcher->timeoutMs = timeoutMs;
+  dispatcher->isThreadLocal = isThreadLocal;
 
-  if (pthread_key_create(&dispatcher->key, destroyDispatcher)) {
-    free(dispatcher);
-    return NULL;
+  if (isThreadLocal) {
+    if (pthread_key_create(&dispatcher->key, destroyDispatcher)) {
+      free(dispatcher);
+      return NULL;
+    }
+  } else {
+    dispatcher->global = createAsyncBulkWriteDispatcher(batchSize, timeoutMs);
+    if (!dispatcher->global) {
+      free(dispatcher);
+      return NULL;
+    }
   }
   return dispatcher;
 }
 
-SAsyncBulkWriteDispatcher* dispatcherThreadLocal(SThreadLocalDispatcher* dispatcher) {
-  SAsyncBulkWriteDispatcher* value = pthread_getspecific(dispatcher->key);
+SAsyncBulkWriteDispatcher* dispatcherAcquire(SDispatcherHolder* holder) {
+  if (!holder->isThreadLocal) {
+    return holder->global;
+  }
+
+  SAsyncBulkWriteDispatcher* value = pthread_getspecific(holder->key);
   if (value) {
     return value;
   }
 
-  value = createAsyncBulkWriteDispatcher(dispatcher->batchSize, dispatcher->timeoutMs);
+  value = createAsyncBulkWriteDispatcher(holder->batchSize, holder->timeoutMs);
   if (value) {
-    pthread_setspecific(dispatcher->key, value);
+    pthread_setspecific(holder->key, value);
     return value;
   }
 
   return NULL;
 }
 
-void destroyThreadLocalDispatcher(SThreadLocalDispatcher* dispatcher) {
-  if (dispatcher) {
-    pthread_key_delete(dispatcher->key);
-    free(dispatcher);
+void destroyDispatcherHolder(SDispatcherHolder* holder) {
+  if (holder) {
+    if (holder->isThreadLocal) {
+      pthread_key_delete(holder->key);
+    } else {
+      destroyAsyncDispatcher(holder->global);
+    }
+    free(holder);
   }
 }

src/client/src/tscSystem.c

@@ -50,7 +50,7 @@ void      *tscRpcCache;            // cache to keep rpc obj
 int32_t    tscNumOfThreads = 1;     // num of rpc threads
 char       tscLogFileName[] = "taoslog";
 int        tscLogFileNum = 10;
-SThreadLocalDispatcher * tscDispatcher = NULL;
+SDispatcherHolder * tscDispatcher = NULL;
 
 static pthread_mutex_t rpcObjMutex; // mutex to protect open the rpc obj concurrently
 static pthread_once_t  tscinit = PTHREAD_ONCE_INIT;
@@ -59,22 +59,17 @@ static pthread_mutex_t setConfMutex = PTHREAD_MUTEX_INITIALIZER;
 // pthread_once can not return result code, so result code is set to a global variable.
 static volatile int tscInitRes = 0;
 
-/**
- * Init the thread local async bulk write dispatcher.
- * 
- * @param batchSize  the batchSize of async bulk write dispatcher.
- * @param timeoutMs  the timeout of batching in milliseconds.
- */
-void tscInitAsyncDispatcher(int32_t batchSize, int32_t timeoutMs) {
-  tscDispatcher = createThreadLocalDispatcher(batchSize, timeoutMs);
+void tscInitAsyncDispatcher(int32_t batchSize, int32_t timeoutMs, bool isThreadLocal) {
+  if (tsAsyncBatchEnable) {
+    tscDispatcher = createDispatcherHolder(batchSize, timeoutMs, isThreadLocal);
+  }
 }
 
-/**
- * Destroy the thread local async bulk write dispatcher.
- */
 void tscDestroyAsyncDispatcher() {
-  destroyThreadLocalDispatcher(tscDispatcher);
-  tscDispatcher = NULL;
+  if (tscDispatcher) {
+    destroyDispatcherHolder(tscDispatcher);
+    tscDispatcher = NULL;
+  }
 }
 
 void tscCheckDiskUsage(void *UNUSED_PARAM(para), void *UNUSED_PARAM(param)) {
@@ -236,9 +231,7 @@ void taos_init_imp(void) {
     tscDebug("starting to initialize client ...");
     tscDebug("Local End Point is:%s", tsLocalEp);
     
-    if (tsAsyncBatchEnable) {
-      tscInitAsyncDispatcher(tsAsyncBatchSize, tsAsyncBatchTimeout);
-    }
+    tscInitAsyncDispatcher(tsAsyncBatchSize, tsAsyncBatchTimeout, tsAsyncBatchThreadLocal);
   }
 
   taosSetCoreDump();

src/common/inc/tglobal.h

@@ -91,8 +91,9 @@ extern float   tsStreamComputDelayRatio;  // the delayed computing ration of the
 extern int32_t tsProjectExecInterval;
 extern int64_t tsMaxRetentWindow;
 extern bool    tsAsyncBatchEnable;
+extern bool    tsAsyncBatchThreadLocal;
 extern int32_t tsAsyncBatchSize;
-extern int64_t tsAsyncBatchTimeout;
+extern int32_t tsAsyncBatchTimeout;
 
 // db parameters in client
 extern int32_t tsCacheBlockSize;

src/common/src/tglobal.c

@@ -127,8 +127,9 @@ int64_t tsMaxRetentWindow = 24 * 3600L;  // maximum time window tolerance
 // The statements will be sent to vnodes no more than `tsAsyncBatchTimeout` milliseconds. But the actual time vnodes
 // received the statements depends on the network quality.
 bool    tsAsyncBatchEnable = true;
+bool    tsAsyncBatchThreadLocal = false;    // if thread local enable, each thread will allocate a dispatcher.
 int32_t tsAsyncBatchSize = 256;
-int64_t tsAsyncBatchTimeout = 5;
+int32_t tsAsyncBatchTimeout = 5;
 
 // the maximum allowed query buffer size during query processing for each data node.
 // -1 no limit (default)
@@ -1842,13 +1843,23 @@ static void doInitGlobalConfig(void) {
   cfg.ptrLength = 0;
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
+  
+  cfg.option = "asyncBatchThreadLocal";
+  cfg.ptr = &tsAsyncBatchThreadLocal;
+  cfg.valType = TAOS_CFG_VTYPE_INT8;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
+  cfg.minValue = 0;
+  cfg.maxValue = 1;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
 
   cfg.option = "asyncBatchSize";
   cfg.ptr = &tsAsyncBatchSize;
   cfg.valType = TAOS_CFG_VTYPE_INT32;
   cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
   cfg.minValue = 1;
-  cfg.maxValue = 65536;
+  cfg.maxValue = 65535;
   cfg.ptrLength = 0;
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
@@ -1858,7 +1869,7 @@ static void doInitGlobalConfig(void) {
   cfg.valType = TAOS_CFG_VTYPE_INT32;
   cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
   cfg.minValue = 1;
-  cfg.maxValue = 65536;
+  cfg.maxValue = 2048;
   cfg.ptrLength = 0;
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);

src/util/inc/tconfig.h

@@ -20,7 +20,7 @@
 extern "C" {
 #endif
 
-#define TSDB_CFG_MAX_NUM    137
+#define TSDB_CFG_MAX_NUM    138
 #define TSDB_CFG_PRINT_LEN  23
 #define TSDB_CFG_OPTION_LEN 24
 #define TSDB_CFG_VALUE_LEN  41