Merge remote-tracking branch 'origin/develop' into feature/os

src/dnode/src/dnodeModule.c

@@ -147,13 +147,13 @@ void dnodeProcessModuleStatus(uint32_t moduleStatus) {
 }
 
 bool dnodeCheckMnodeStarting() {
-  if (tsModuleStatus & TSDB_MOD_MNODE) return false;
+  if (tsModuleStatus & (1 << TSDB_MOD_MNODE)) return false;
 
   SDMMnodeInfos *mnodes = dnodeGetMnodeInfos();
   for (int32_t i = 0; i < mnodes->nodeNum; ++i) {
     SDMMnodeInfo *node = &mnodes->nodeInfos[i];
     if (node->nodeId == dnodeGetDnodeId()) {
-      uint32_t moduleStatus = tsModuleStatus | (1 << TSDB_MOD_MNODE);;
+      uint32_t moduleStatus = tsModuleStatus | (1 << TSDB_MOD_MNODE);
       dInfo("start mnode module, module status:%d, new status:%d", tsModuleStatus, moduleStatus);
       dnodeProcessModuleStatus(moduleStatus);
       return true;

src/mnode/src/mnodeTable.c

@@ -2093,11 +2093,11 @@ static int32_t mnodeDoGetChildTableMeta(SMnodeMsg *pMsg, STableMetaMsg *pMeta) {
   pMeta->precision = pDb->cfg.precision;
   pMeta->tableType = pTable->info.type;
   tstrncpy(pMeta->tableId, pTable->info.tableId, TSDB_TABLE_FNAME_LEN);
-  if (pTable->superTable) {
+  if (pTable->superTable != NULL) {
     tstrncpy(pMeta->sTableId, pTable->superTable->info.tableId, TSDB_TABLE_FNAME_LEN);
   }
 
-  if (pTable->info.type == TSDB_CHILD_TABLE) {
+  if (pTable->info.type == TSDB_CHILD_TABLE && pTable->superTable != NULL) {
     pMeta->sversion     = htons(pTable->superTable->sversion);
     pMeta->tversion     = htons(pTable->superTable->tversion);
     pMeta->numOfTags    = (int8_t)pTable->superTable->numOfTags;

src/os/src/detail/osDir.c

@@ -126,12 +126,13 @@ int32_t taosCompressFile(char *srcFileName, char *destFileName) {
   dstFp = gzdopen(fd, "wb6f");
   if (dstFp == NULL) {
     ret = -3;
+    close(fd);
     goto cmp_end;
   }
 
   while (!feof(srcFp)) {
     len = (int32_t)fread(data, 1, COMPRESS_STEP_SIZE, srcFp);
-    gzwrite(dstFp, data, len);
+    (void)gzwrite(dstFp, data, len);
   }
 
 cmp_end:

src/plugins/http/src/httpParser.c

@@ -153,10 +153,10 @@ static int32_t httpOnRequestLine(HttpParser *pParser, char *method, char *target
   for (int32_t i = 0; i < HTTP_MAX_URL; i++) {
     char *pSeek = strchr(pStart, '/');
     if (pSeek == NULL) {
-      httpAppendString(pParser->path + i, pStart, strlen(pStart));
+      (void)httpAppendString(pParser->path + i, pStart, strlen(pStart));
       break;
     } else {
-      httpAppendString(pParser->path + i, pStart, (int32_t)(pSeek - pStart));
+      (void)httpAppendString(pParser->path + i, pStart, (int32_t)(pSeek - pStart));
     }
     pStart = pSeek + 1;
   }
@@ -485,11 +485,11 @@ void httpClearParser(HttpParser *parser) {
 }
 
 void httpDestroyParser(HttpParser *parser) {
+  if (!parser) return;
+
   HttpContext *pContext = parser->pContext;
   httpTrace("context:%p, fd:%d, destroy parser", pContext, pContext->fd);
 
-  if (!parser) return;
-
   free(parser->method);         parser->method          = NULL;
   free(parser->target);         parser->target          = NULL;
   free(parser->version);        parser->version         = NULL;
@@ -684,12 +684,13 @@ static int32_t httpParserOnVersion(HttpParser *parser, HTTP_PARSER_STATE state,
       break;
     }
 
-    if (c!='0' && c!='1') {
+    if (c != '0' && c != '1' && c != '2') {
       httpError("context:%p, fd:%d, parser state:%d, unexpected char:[%c]%02x", pContext, pContext->fd, state, c, c);
       ok = -1;
       httpOnError(parser, 400, TSDB_CODE_HTTP_PARSE_VERSION_FAILED);
       break;
     }
+
     if (httpAppendString(&parser->str, &c, 1)) {
       httpError("context:%p, fd:%d, parser state:%d, char:[%c]%02x, oom", pContext, pContext->fd, state, c, c);
       ok = -1;
@@ -697,10 +698,14 @@ static int32_t httpParserOnVersion(HttpParser *parser, HTTP_PARSER_STATE state,
       break;
     }
 
-    if (c == '0') parser->httpVersion = HTTP_VERSION_10;
-    else if (c == '1') parser->httpVersion = HTTP_VERSION_11;
-    else if (c == '2') parser->httpVersion = HTTP_VERSION_12;
-    else parser->httpVersion = HTTP_INVALID_VERSION;
+    if (c == '0')
+      parser->httpVersion = HTTP_VERSION_10;
+    else if (c == '1')
+      parser->httpVersion = HTTP_VERSION_11;
+    else if (c == '2')
+      parser->httpVersion = HTTP_VERSION_12;
+    else {
+    }
 
     parser->version = strdup(parser->str.str);
     if (!parser->version) {

src/sync/inc/syncInt.h

@@ -118,7 +118,7 @@ typedef struct SsyncPeer {
   uint32_t ip;
   uint16_t port;
   char     fqdn[TSDB_FQDN_LEN];   // peer ip string
-  char        id[TSDB_EP_LEN+16];   // peer vgId + end point
+  char     id[TSDB_EP_LEN + 32];  // peer vgId + end point
   int8_t   role;
   int8_t   sstatus;   // sync status
   uint64_t version;
@@ -127,11 +127,11 @@ typedef struct SsyncPeer {
   int      peerFd;          // forward FD
   int      numOfRetrieves;  // number of retrieves tried
   int      fileChanged;     // a flag to indicate file is changed during retrieving process
-  void       *timer;
-  void       *pConn;
+  void *   timer;
+  void *   pConn;
   int      notifyFd;
   int      watchNum;
-  int        *watchFd;
+  int *    watchFd;
   int8_t   refCount;  // reference count
   struct   SSyncNode *pSyncNode;
 } SSyncPeer;
@@ -171,7 +171,6 @@ void  syncBroadcastStatus(SSyncNode *pNode);
 void  syncAddPeerRef(SSyncPeer *pPeer);
 int   syncDecPeerRef(SSyncPeer *pPeer);
 
-
 #ifdef __cplusplus
 }
 #endif

src/sync/inc/taosTcpPool.h

@@ -38,7 +38,6 @@ void       taosCloseTcpThreadPool(ttpool_h);
 void      *taosAllocateTcpConn(void *, void *ahandle, int connFd);
 void       taosFreeTcpConn(void *);
 
-
 #ifdef __cplusplus
 }
 #endif

src/sync/src/syncMain.c

@@ -134,7 +134,7 @@ void syncCleanUp() {
 void *syncStart(const SSyncInfo *pInfo) {
   const SSyncCfg *pCfg = &pInfo->syncCfg;
 
-  SSyncNode *pNode = (SSyncNode *) calloc(sizeof(SSyncNode), 1);
+  SSyncNode *pNode = (SSyncNode *)calloc(sizeof(SSyncNode), 1);
   if (pNode == NULL) {
     sError("no memory to allocate syncNode");
     terrno = TAOS_SYSTEM_ERROR(errno);
@@ -178,9 +178,10 @@ void *syncStart(const SSyncInfo *pInfo) {
 
   nodeVersion = pInfo->version;  // set the initial version
   nodeRole = (pNode->replica > 1) ? TAOS_SYNC_ROLE_UNSYNCED : TAOS_SYNC_ROLE_MASTER;
-  sInfo("vgId:%d, %d replicas are configured, quorum:%d role:%s", pNode->vgId, pNode->replica, pNode->quorum, syncRole[nodeRole]);
+  sInfo("vgId:%d, %d replicas are configured, quorum:%d role:%s", pNode->vgId, pNode->replica, pNode->quorum,
+        syncRole[nodeRole]);
 
-  pNode->pSyncFwds = calloc(sizeof(SSyncFwds) + tsMaxFwdInfo*sizeof(SFwdInfo), 1);
+  pNode->pSyncFwds = calloc(sizeof(SSyncFwds) + tsMaxFwdInfo * sizeof(SFwdInfo), 1);
   if (pNode->pSyncFwds == NULL) {
     sError("vgId:%d, no memory to allocate syncFwds", pNode->vgId);
     terrno = TAOS_SYSTEM_ERROR(errno);
@@ -443,9 +444,7 @@ static void syncAddArbitrator(SSyncNode *pNode) {
   pNode->peerInfo[TAOS_SYNC_MAX_REPLICA] = syncAddPeer(pNode, &nodeInfo);
 }
 
-static void syncAddNodeRef(SSyncNode *pNode) {
-  atomic_add_fetch_8(&pNode->refCount, 1);
-}
+static void syncAddNodeRef(SSyncNode *pNode) { atomic_add_fetch_8(&pNode->refCount, 1); }
 
 static void syncDecNodeRef(SSyncNode *pNode) {
   if (atomic_sub_fetch_8(&pNode->refCount, 1) == 0) {
@@ -456,9 +455,7 @@ static void syncDecNodeRef(SSyncNode *pNode) {
   }
 }
 
-void syncAddPeerRef(SSyncPeer *pPeer) {
-  atomic_add_fetch_8(&pPeer->refCount, 1);
-}
+void syncAddPeerRef(SSyncPeer *pPeer) { atomic_add_fetch_8(&pPeer->refCount, 1); }
 
 int syncDecPeerRef(SSyncPeer *pPeer) {
   if (atomic_sub_fetch_8(&pPeer->refCount, 1) == 0) {
@@ -501,6 +498,7 @@ static SSyncPeer *syncAddPeer(SSyncNode *pNode, const SNodeInfo *pInfo) {
   tstrncpy(pPeer->fqdn, pInfo->nodeFqdn, sizeof(pPeer->fqdn));
   pPeer->ip = ip;
   pPeer->port = pInfo->nodePort;
+  pPeer->fqdn[sizeof(pPeer->fqdn) - 1] = 0;
   snprintf(pPeer->id, sizeof(pPeer->id), "vgId:%d peer:%s:%d", pNode->vgId, pPeer->fqdn, pPeer->port);
 
   pPeer->peerFd = -1;
@@ -573,10 +571,10 @@ static void syncChooseMaster(SSyncNode *pNode) {
     replica = pNode->replica + 1;
   }
 
-  if (index < 0 && onlineNum > replica/2.0) {
+  if (index < 0 && onlineNum > replica / 2.0) {
     // over half of nodes are online
     for (int i = 0; i < pNode->replica; ++i) {
-      //slave with highest version shall be master
+      // slave with highest version shall be master
       pPeer = pNode->peerInfo[i];
       if (pPeer->role == TAOS_SYNC_ROLE_SLAVE || pPeer->role == TAOS_SYNC_ROLE_MASTER) {
         if (index < 0 || pPeer->version > pNode->peerInfo[index]->version) {
@@ -622,7 +620,7 @@ static SSyncPeer *syncCheckMaster(SSyncNode *pNode) {
   if (onlineNum <= replica * 0.5) {
     if (nodeRole != TAOS_SYNC_ROLE_UNSYNCED) {
       nodeRole = TAOS_SYNC_ROLE_UNSYNCED;
-      pNode->peerInfo[pNode->selfIndex]->role = nodeRole;
+      // pNode->peerInfo[pNode->selfIndex]->role = nodeRole;
       (*pNode->notifyRole)(pNode->ahandle, nodeRole);
       sInfo("vgId:%d, change to unsynced state, online:%d replica:%d", pNode->vgId, onlineNum, replica);
     }
@@ -671,7 +669,7 @@ static void syncCheckRole(SSyncPeer *pPeer, SPeerStatus peersStatus[], int8_t ne
   int8_t     selfOldRole = nodeRole;
   int8_t     i, syncRequired = 0;
 
-  pNode->peerInfo[pNode->selfIndex]->version = nodeVersion;
+  // pNode->peerInfo[pNode->selfIndex]->version = nodeVersion;
   pPeer->role = newRole;
 
   sDebug("%s, own role:%s, new peer role:%s", pPeer->id, syncRole[nodeRole], syncRole[pPeer->role]);
@@ -877,8 +875,6 @@ static void syncProcessForwardFromPeer(char *cont, SSyncPeer *pPeer) {
       sError("%s, forward discarded, ver:%" PRIu64, pPeer->id, pHead->version);
     }
   }
-
-  return;
 }
 
 static void syncProcessPeersStatusMsg(char *cont, SSyncPeer *pPeer) {
@@ -923,7 +919,7 @@ static int syncReadPeerMsg(SSyncPeer *pPeer, SSyncHead *pHead, char *cont) {
 static int syncProcessPeerMsg(void *param, void *buffer) {
   SSyncPeer *pPeer = param;
   SSyncHead  head;
-  char *     cont = (char *)buffer;
+  char *     cont = buffer;
 
   SSyncNode *pNode = pPeer->pSyncNode;
   pthread_mutex_lock(&(pNode->mutex));

src/sync/src/syncRestore.c

@@ -23,9 +23,9 @@
 #include "tsync.h"
 #include "syncInt.h"
 
-static void syncRemoveExtraFile(SSyncPeer *pPeer, uint32_t sindex, uint32_t eindex) {
-  char       name[TSDB_FILENAME_LEN*2] = {0};
-  char       fname[TSDB_FILENAME_LEN*3] = {0};
+static void syncRemoveExtraFile(SSyncPeer *pPeer, int32_t sindex, int32_t eindex) {
+  char       name[TSDB_FILENAME_LEN * 2] = {0};
+  char       fname[TSDB_FILENAME_LEN * 3] = {0};
   uint32_t   magic;
   uint64_t   fversion;
   int64_t    size;
@@ -40,7 +40,7 @@ static void syncRemoveExtraFile(SSyncPeer *pPeer, uint32_t sindex, uint32_t eind
     if (magic == 0) break;
 
     snprintf(fname, sizeof(fname), "%s/%s", pNode->path, name);
-    remove(fname);
+    (void)remove(fname);
     sDebug("%s, %s is removed", pPeer->id, fname);
 
     index++;
@@ -62,34 +62,35 @@ static int syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
   while (1) {
     // read file info
     int ret = taosReadMsg(pPeer->syncFd, &(minfo), sizeof(minfo));
-    if (ret < 0 ) break;
+    if (ret < 0) break;
 
     // if no more file from master, break;
     if (minfo.name[0] == 0 || minfo.magic == 0) {
       sDebug("%s, no more files to restore", pPeer->id);
 
       // remove extra files after the current index
-      syncRemoveExtraFile(pPeer, sinfo.index+1, TAOS_SYNC_MAX_INDEX);
+      syncRemoveExtraFile(pPeer, sinfo.index + 1, TAOS_SYNC_MAX_INDEX);
       code = 0;
       break;
     }
 
     // remove extra files on slave between the current and last index
-    syncRemoveExtraFile(pPeer, pindex+1, minfo.index-1);
+    syncRemoveExtraFile(pPeer, pindex + 1, minfo.index - 1);
     pindex = minfo.index;
 
     // check the file info
     sinfo = minfo;
     sDebug("%s, get file info:%s", pPeer->id, minfo.name);
-    sinfo.magic = (*pNode->getFileInfo)(pNode->ahandle, sinfo.name, &sinfo.index, TAOS_SYNC_MAX_INDEX, &sinfo.size, &sinfo.fversion);
+    sinfo.magic = (*pNode->getFileInfo)(pNode->ahandle, sinfo.name, &sinfo.index, TAOS_SYNC_MAX_INDEX, &sinfo.size,
+                                        &sinfo.fversion);
 
     // if file not there or magic is not the same, file shall be synced
     memset(&fileAck, 0, sizeof(fileAck));
-    fileAck.sync = (sinfo.magic != minfo.magic || sinfo.name[0] == 0) ? 1:0;
+    fileAck.sync = (sinfo.magic != minfo.magic || sinfo.name[0] == 0) ? 1 : 0;
 
     // send file ack
     ret = taosWriteMsg(pPeer->syncFd, &(fileAck), sizeof(fileAck));
-    if (ret <0)  break;
+    if (ret < 0) break;
 
     // if sync is not required, continue
     if (fileAck.sync == 0) {
@@ -99,10 +100,11 @@ static int syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
 
     // if sync is required, open file, receive from master, and write to file
     // get the full path to file
+    minfo.name[sizeof(minfo.name) - 1] = 0;
     snprintf(name, sizeof(name), "%s/%s", pNode->path, minfo.name);
 
     int dfd = open(name, O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU | S_IRWXG | S_IRWXO);
-    if ( dfd < 0 ) {
+    if (dfd < 0) {
       sError("%s, failed to open file:%s", pPeer->id, name);
       break;
     }
@@ -110,10 +112,9 @@ static int syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
     ret = taosCopyFds(pPeer->syncFd, dfd, minfo.size);
     fsync(dfd);
     close(dfd);
-    if (ret<0) break;
+    if (ret < 0) break;
 
     sDebug("%s, %s is received, size:%" PRId64, pPeer->id, minfo.name, minfo.size);
-
   }
 
   if (code == 0 && (minfo.fversion != sinfo.fversion)) {
@@ -140,18 +141,21 @@ static int syncRestoreWal(SSyncPeer *pPeer) {
 
   while (1) {
     ret = taosReadMsg(pPeer->syncFd, pHead, sizeof(SWalHead));
-    if (ret <0)  break;
+    if (ret < 0) break;
 
-    if (pHead->len == 0) {code = 0; break;}  // wal sync over
+    if (pHead->len == 0) {
+      code = 0;
+      break;
+    }  // wal sync over
 
     ret = taosReadMsg(pPeer->syncFd, pHead->cont, pHead->len);
-    if (ret <0)  break;
+    if (ret < 0) break;
 
     sDebug("%s, restore a record, ver:%" PRIu64, pPeer->id, pHead->version);
     (*pNode->writeToCache)(pNode->ahandle, pHead, TAOS_QTYPE_WAL);
   }
 
-  if (code<0) {
+  if (code < 0) {
     sError("%s, failed to restore wal(%s)", pPeer->id, strerror(errno));
   }
 
@@ -159,10 +163,9 @@ static int syncRestoreWal(SSyncPeer *pPeer) {
   return code;
 }
 
-static char *syncProcessOneBufferedFwd(SSyncPeer *pPeer, char *offset)
-{
+static char *syncProcessOneBufferedFwd(SSyncPeer *pPeer, char *offset) {
   SSyncNode *pNode = pPeer->pSyncNode;
-  SWalHead  *pHead = (SWalHead *) offset;
+  SWalHead * pHead = (SWalHead *)offset;
 
   (*pNode->writeToCache)(pNode->ahandle, pHead, TAOS_QTYPE_FWD);
   offset += pHead->len + sizeof(SWalHead);
@@ -171,7 +174,7 @@ static char *syncProcessOneBufferedFwd(SSyncPeer *pPeer, char *offset)
 }
 
 static int syncProcessBufferedFwd(SSyncPeer *pPeer) {
-  SSyncNode   *pNode = pPeer->pSyncNode;
+  SSyncNode *  pNode = pPeer->pSyncNode;
   SRecvBuffer *pRecv = pNode->pRecv;
   int          forwards = 0;
 
@@ -199,7 +202,7 @@ static int syncProcessBufferedFwd(SSyncPeer *pPeer) {
 }
 
 int syncSaveIntoBuffer(SSyncPeer *pPeer, SWalHead *pHead) {
-  SSyncNode   *pNode = pPeer->pSyncNode;
+  SSyncNode *  pNode = pPeer->pSyncNode;
   SRecvBuffer *pRecv = pNode->pRecv;
 
   if (pRecv == NULL) return -1;
@@ -261,7 +264,7 @@ static int syncRestoreDataStepByStep(SSyncPeer *pPeer) {
 
   // if code > 0, data file is changed, notify app, and pass the version
   if (code > 0 && pNode->notifyFileSynced) {
-    if ( (*pNode->notifyFileSynced)(pNode->ahandle, fversion) < 0 ) {
+    if ((*pNode->notifyFileSynced)(pNode->ahandle, fversion) < 0) {
       sError("%s, app not in ready state", pPeer->id);
       return -1;
     }
@@ -297,7 +300,7 @@ void *syncRestoreData(void *param) {
   if (syncOpenRecvBuffer(pNode) < 0) {
     sError("%s, failed to allocate recv buffer", pPeer->id);
   } else {
-    if ( syncRestoreDataStepByStep(pPeer) == 0) {
+    if (syncRestoreDataStepByStep(pPeer) == 0) {
       sInfo("%s, it is synced successfully", pPeer->id);
       nodeRole = TAOS_SYNC_ROLE_SLAVE;
       syncBroadcastStatus(pNode);
@@ -311,7 +314,7 @@ void *syncRestoreData(void *param) {
   (*pNode->notifyRole)(pNode->ahandle, nodeRole);
 
   nodeSStatus = TAOS_SYNC_STATUS_INIT;
-  taosClose(pPeer->syncFd)
+  taosClose(pPeer->syncFd);
   syncCloseRecvBuffer(pNode);
   __sync_fetch_and_sub(&tsSyncNum, 1);
   syncDecPeerRef(pPeer);

src/sync/src/syncRetrieve.c

@@ -38,13 +38,13 @@ static int syncAddIntoWatchList(SSyncPeer *pPeer, char *name) {
       return -1;
     }
 
-    if (pPeer->watchFd == NULL) pPeer->watchFd = malloc(sizeof(int)*tsMaxWatchFiles);
+    if (pPeer->watchFd == NULL) pPeer->watchFd = malloc(sizeof(int) * tsMaxWatchFiles);
     if (pPeer->watchFd == NULL) {
       sError("%s, failed to allocate watchFd", pPeer->id);
       return -1;
     }
 
-    memset(pPeer->watchFd, -1, sizeof(int)*tsMaxWatchFiles);
+    memset(pPeer->watchFd, -1, sizeof(int) * tsMaxWatchFiles);
   }
 
   int *wd = pPeer->watchFd + pPeer->watchNum;
@@ -64,7 +64,7 @@ static int syncAddIntoWatchList(SSyncPeer *pPeer, char *name) {
     sDebug("%s, monitor %s, wd:%d watchNum:%d", pPeer->id, name, *wd, pPeer->watchNum);
   }
 
-  pPeer->watchNum = (pPeer->watchNum +1) % tsMaxWatchFiles;
+  pPeer->watchNum = (pPeer->watchNum + 1) % tsMaxWatchFiles;
 
   return 0;
 }
@@ -84,7 +84,7 @@ static int syncAreFilesModified(SSyncPeer *pPeer) {
     const struct inotify_event *event;
     char *ptr;
     for (ptr = buf; ptr < buf + len; ptr += sizeof(struct inotify_event) + event->len) {
-      event = (const struct inotify_event *) ptr; 
+      event = (const struct inotify_event *)ptr;
       if ((event->mask & IN_MODIFY) || (event->mask & IN_DELETE)) {
         sDebug("%s, processed file is changed", pPeer->id);
         pPeer->fileChanged = 1;
@@ -98,7 +98,7 @@ static int syncAreFilesModified(SSyncPeer *pPeer) {
 }
 
 static int syncRetrieveFile(SSyncPeer *pPeer) {
-  SSyncNode * pNode = pPeer->pSyncNode;
+  SSyncNode *pNode = pPeer->pSyncNode;
   SFileInfo  fileInfo;
   SFileAck   fileAck;
   int        code = -1;
@@ -110,17 +110,19 @@ static int syncRetrieveFile(SSyncPeer *pPeer) {
   while (1) {
     // retrieve file info
     fileInfo.name[0] = 0;
-    fileInfo.magic = (*pNode->getFileInfo)(pNode->ahandle, fileInfo.name, &fileInfo.index, TAOS_SYNC_MAX_INDEX, &fileInfo.size, &fileInfo.fversion);   
-    //fileInfo.size = htonl(size);
+    fileInfo.magic = (*pNode->getFileInfo)(pNode->ahandle, fileInfo.name, &fileInfo.index, TAOS_SYNC_MAX_INDEX,
+                                           &fileInfo.size, &fileInfo.fversion);
+    // fileInfo.size = htonl(size);
 
     // send the file info
     int32_t ret = taosWriteMsg(pPeer->syncFd, &(fileInfo), sizeof(fileInfo));
-    if (ret < 0 ) break;
+    if (ret < 0) break;
 
     // if no file anymore, break
     if (fileInfo.magic == 0 || fileInfo.name[0] == 0) {
       sDebug("%s, no more files to sync", pPeer->id);
-      code = 0; break; 
+      code = 0;
+      break;
     }
 
     // wait for the ack from peer
@@ -134,7 +136,7 @@ static int syncRetrieveFile(SSyncPeer *pPeer) {
     snprintf(name, sizeof(name), "%s/%s", pNode->path, fileInfo.name);
 
     // add the file into watch list
-    if ( syncAddIntoWatchList(pPeer, name) <0) break;
+    if (syncAddIntoWatchList(pPeer, name) < 0) break;
 
     // if sync is not required, continue
     if (fileAck.sync == 0) {
@@ -201,13 +203,13 @@ static int syncMonitorLastWal(SSyncPeer *pPeer, char *name) {
     return -1;
   }
 
-  if (pPeer->watchFd == NULL) pPeer->watchFd = malloc(sizeof(int)*tsMaxWatchFiles);
+  if (pPeer->watchFd == NULL) pPeer->watchFd = malloc(sizeof(int) * tsMaxWatchFiles);
   if (pPeer->watchFd == NULL) {
     sError("%s, failed to allocate watchFd", pPeer->id);
     return -1;
   }
 
-  memset(pPeer->watchFd, -1, sizeof(int)*tsMaxWatchFiles);
+  memset(pPeer->watchFd, -1, sizeof(int) * tsMaxWatchFiles);
   int *wd = pPeer->watchFd;
 
   *wd = inotify_add_watch(pPeer->notifyFd, name, IN_MODIFY | IN_CLOSE_WRITE);
@@ -219,7 +221,7 @@ static int syncMonitorLastWal(SSyncPeer *pPeer, char *name) {
   return 0;
 }
 
-static uint32_t syncCheckLastWalChanges(SSyncPeer *pPeer, uint32_t *pEvent) {
+static int32_t syncCheckLastWalChanges(SSyncPeer *pPeer, uint32_t *pEvent) {
   char buf[2048];
   int  len = read(pPeer->notifyFd, buf, sizeof(buf));
   if (len < 0 && errno != EAGAIN) {
@@ -231,26 +233,29 @@ static uint32_t syncCheckLastWalChanges(SSyncPeer *pPeer, uint32_t *pEvent) {
 
   struct inotify_event *event;
   for (char *ptr = buf; ptr < buf + len; ptr += sizeof(struct inotify_event) + event->len) {
-    event = (struct inotify_event *) ptr;
+    event = (struct inotify_event *)ptr;
     if (event->mask & IN_MODIFY) *pEvent = *pEvent | IN_MODIFY;
     if (event->mask & IN_CLOSE_WRITE) *pEvent = *pEvent | IN_CLOSE_WRITE;
   }
 
-  if (pEvent != 0)
-    sDebug("%s, last wal event:0x%x", pPeer->id, *pEvent);
+  if (pEvent != 0) sDebug("%s, last wal event:0x%x", pPeer->id, *pEvent);
 
   return 0;
 }
 
 static int syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion, int64_t offset, uint32_t *pEvent) {
-  SWalHead  *pHead = (SWalHead *) malloc(640000);
+  SWalHead *pHead = malloc(640000);
   int       code = -1;
   int32_t   bytes = 0;
   int       sfd;
 
   sfd = open(name, O_RDONLY);
-  if (sfd < 0) return -1;
-  lseek(sfd, offset, SEEK_SET);
+  if (sfd < 0) {
+    free(pHead);
+    return -1;
+  }
+
+  (void)lseek(sfd, offset, SEEK_SET);
   sDebug("%s, retrieve last wal, offset:%" PRId64 " fversion:%" PRIu64, pPeer->id, offset, fversion);
 
   while (1) {
@@ -263,7 +268,7 @@ static int syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion,
 
     sDebug("%s, last wal is forwarded, ver:%" PRIu64, pPeer->id, pHead->version);
     int ret = taosWriteMsg(pPeer->syncFd, pHead, wsize);
-    if ( ret != wsize ) break;
+    if (ret != wsize) break;
     pPeer->sversion = pHead->version;
 
     bytes += wsize;
@@ -276,7 +281,7 @@ static int syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion,
   }
 
   free(pHead);
-  taosClose(sfd); 
+  close(sfd);
 
   if (code == 0) return bytes;
   return -1;
@@ -300,14 +305,14 @@ static int syncProcessLastWal(SSyncPeer *pPeer, char *wname, uint32_t index) {
     sDebug("%s, start to retrieve last wal:%s", pPeer->id, fname);
 
     // monitor last wal
-    if (syncMonitorLastWal(pPeer, fname) <0) break;
+    if (syncMonitorLastWal(pPeer, fname) < 0) break;
 
     while (1) {
       int32_t bytes = syncRetrieveLastWal(pPeer, fname, fversion, offset, &event);
       if (bytes < 0) break;
 
       // check file changes
-      if (syncCheckLastWalChanges(pPeer, &event) <0) break;
+      if (syncCheckLastWalChanges(pPeer, &event) < 0) break;
 
       // if file is not updated or updated once, set the fversion and sstatus
       if (((event & IN_MODIFY) == 0) || once) {
@@ -403,7 +408,7 @@ static int syncRetrieveWal(SSyncPeer *pPeer) {
 
     code = taosTSendFile(pPeer->syncFd, sfd, NULL, size);
     close(sfd);
-    if (code <0) break;
+    if (code < 0) break;
 
     index++;
 
@@ -433,7 +438,7 @@ static int syncRetrieveDataStepByStep(SSyncPeer *pPeer) {
   tstrncpy(firstPkt.fqdn, tsNodeFqdn, sizeof(firstPkt.fqdn));
   firstPkt.port = tsSyncPort;
 
-  if (write(pPeer->syncFd, (char *) &firstPkt, sizeof(firstPkt)) < 0) {
+  if (write(pPeer->syncFd, (char *)&firstPkt, sizeof(firstPkt)) < 0) {
     sError("%s, failed to send syncCmd", pPeer->id);
     return -1;
   }
@@ -447,8 +452,7 @@ static int syncRetrieveDataStepByStep(SSyncPeer *pPeer) {
   }
 
   // if no files are synced, there must be wal to sync, sversion must be larger than one
-  if (pPeer->sversion == 0) 
-    pPeer->sversion = 1;
+  if (pPeer->sversion == 0) pPeer->sversion = 1;
 
   sDebug("%s, start to retrieve wal", pPeer->id);
   if (syncRetrieveWal(pPeer) < 0) {
@@ -460,7 +464,7 @@ static int syncRetrieveDataStepByStep(SSyncPeer *pPeer) {
 }
 
 void *syncRetrieveData(void *param) {
-  SSyncPeer * pPeer = (SSyncPeer *)param;
+  SSyncPeer *pPeer = (SSyncPeer *)param;
   SSyncNode *pNode = pPeer->pSyncNode;
   taosBlockSIGPIPE();
 
@@ -486,8 +490,7 @@ void *syncRetrieveData(void *param) {
       (*pNode->notifyFlowCtrl)(pNode->ahandle, 4 << (pPeer->numOfRetrieves - 2));
   } else {
     pPeer->numOfRetrieves = 0;
-    if (pNode->notifyFlowCtrl)
-      (*pNode->notifyFlowCtrl)(pNode->ahandle, 0);
+    if (pNode->notifyFlowCtrl) (*pNode->notifyFlowCtrl)(pNode->ahandle, 0);
   }
 
   pPeer->fileChanged = 0;

src/sync/src/taosTcpPool.c

@@ -45,8 +45,8 @@ typedef struct {
 
 static void *taosAcceptPeerTcpConnection(void *argv);
 static void *taosProcessTcpData(void *param);
+static void  taosStopPoolThread(SThreadObj *pThread);
 static SThreadObj *taosGetTcpThread(SPoolObj *pPool);
-static void taosStopPoolThread(SThreadObj* pThread);
 
 void *taosOpenTcpThreadPool(SPoolInfo *pInfo) {
   pthread_attr_t thattr;
@@ -59,7 +59,7 @@ void *taosOpenTcpThreadPool(SPoolInfo *pInfo) {
 
   pPool->info = *pInfo;
 
-  pPool->pThread = (SThreadObj **) calloc(sizeof(SThreadObj *), pInfo->numOfThreads);
+  pPool->pThread = (SThreadObj **)calloc(sizeof(SThreadObj *), pInfo->numOfThreads);
   if (pPool->pThread == NULL) {
     uError("TCP server, no enough memory");
     free(pPool);
@@ -68,17 +68,19 @@ void *taosOpenTcpThreadPool(SPoolInfo *pInfo) {
 
   pPool->acceptFd = taosOpenTcpServerSocket(pInfo->serverIp, pInfo->port);
   if (pPool->acceptFd < 0) {
-    free(pPool->pThread); free(pPool);
+    free(pPool->pThread);
+    free(pPool);
     uError("failed to create TCP server socket, port:%d (%s)", pInfo->port, strerror(errno));
     return NULL;
   }
 
   pthread_attr_init(&thattr);
   pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
-  if (pthread_create(&(pPool->thread), &thattr, (void *) taosAcceptPeerTcpConnection, pPool) != 0) {
+  if (pthread_create(&(pPool->thread), &thattr, (void *)taosAcceptPeerTcpConnection, pPool) != 0) {
     uError("TCP server, failed to create accept thread, reason:%s", strerror(errno));
     close(pPool->acceptFd);
-    free(pPool->pThread); free(pPool);
+    free(pPool->pThread);
+    free(pPool);
     return NULL;
   }
 
@@ -89,7 +91,7 @@ void *taosOpenTcpThreadPool(SPoolInfo *pInfo) {
 }
 
 void taosCloseTcpThreadPool(void *param) {
-  SPoolObj    *pPool = (SPoolObj *)param;
+  SPoolObj *  pPool = (SPoolObj *)param;
   SThreadObj *pThread;
 
   shutdown(pPool->acceptFd, SHUT_RD);
@@ -100,16 +102,17 @@ void taosCloseTcpThreadPool(void *param) {
     if (pThread) taosStopPoolThread(pThread);
   }
 
+  uDebug("%p TCP pool is closed", pPool);
+
   taosTFree(pPool->pThread);
   free(pPool);
-  uDebug("%p TCP pool is closed", pPool);
 }
 
 void *taosAllocateTcpConn(void *param, void *pPeer, int connFd) {
   struct epoll_event event;
   SPoolObj *pPool = (SPoolObj *)param;
 
-  SConnObj *pConn = (SConnObj *) calloc(sizeof(SConnObj), 1);
+  SConnObj *pConn = (SConnObj *)calloc(sizeof(SConnObj), 1);
   if (pConn == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     return NULL;
@@ -153,8 +156,8 @@ void taosFreeTcpConn(void *param) {
 
 static void taosProcessBrokenLink(SConnObj *pConn) {
   SThreadObj *pThread = pConn->pThread;
-  SPoolObj   *pPool = pThread->pPool;
-  SPoolInfo  *pInfo = &pPool->info;
+  SPoolObj *  pPool = pThread->pPool;
+  SPoolInfo * pInfo = &pPool->info;
 
   if (pConn->closedByApp == 0) shutdown(pConn->fd, SHUT_WR);
   (*pInfo->processBrokenLink)(pConn->ahandle);
@@ -169,10 +172,10 @@ static void taosProcessBrokenLink(SConnObj *pConn) {
 #define maxEvents 10
 
 static void *taosProcessTcpData(void *param) {
-  SThreadObj     *pThread = (SThreadObj *) param;
-  SPoolObj       *pPool = pThread->pPool;
-  SPoolInfo      *pInfo = &pPool->info;
-  SConnObj       *pConn = NULL;
+  SThreadObj *pThread = (SThreadObj *)param;
+  SPoolObj *  pPool = pThread->pPool;
+  SPoolInfo * pInfo = &pPool->info;
+  SConnObj *  pConn = NULL;
   struct epoll_event events[maxEvents];
 
   void *buffer = malloc(pInfo->bufferSize);
@@ -219,15 +222,16 @@ static void *taosProcessTcpData(void *param) {
     }
   }
 
+  uDebug("%p TCP epoll thread exits", pThread);
+
   close(pThread->pollFd);
   free(pThread);
   free(buffer);
-  uDebug("%p TCP epoll thread exits", pThread);
   return NULL;
 }
 
 static void *taosAcceptPeerTcpConnection(void *argv) {
-  SPoolObj   *pPool = (SPoolObj *)argv;
+  SPoolObj * pPool = (SPoolObj *)argv;
   SPoolInfo *pInfo = &pPool->info;
 
   taosBlockSIGPIPE();
@@ -235,7 +239,7 @@ static void *taosAcceptPeerTcpConnection(void *argv) {
   while (1) {
     struct sockaddr_in clientAddr;
     socklen_t addrlen = sizeof(clientAddr);
-    int connFd = accept(pPool->acceptFd, (struct sockaddr *) &clientAddr, &addrlen);
+    int connFd = accept(pPool->acceptFd, (struct sockaddr *)&clientAddr, &addrlen);
     if (connFd < 0) {
       if (errno == EINVAL) {
         uDebug("%p TCP server accept is exiting...", pPool);
@@ -246,7 +250,7 @@ static void *taosAcceptPeerTcpConnection(void *argv) {
       }
     }
 
-    //uDebug("TCP connection from: 0x%x:%d", clientAddr.sin_addr.s_addr, clientAddr.sin_port); 
+    // uDebug("TCP connection from: 0x%x:%d", clientAddr.sin_addr.s_addr, clientAddr.sin_port);
     taosKeepTcpAlive(connFd);
     (*pInfo->processIncomingConn)(connFd, clientAddr.sin_addr.s_addr);
   }
@@ -260,7 +264,7 @@ static SThreadObj *taosGetTcpThread(SPoolObj *pPool) {
 
   if (pThread) return pThread;
 
-  pThread = (SThreadObj *) calloc(1, sizeof(SThreadObj));
+  pThread = (SThreadObj *)calloc(1, sizeof(SThreadObj));
   if (pThread == NULL) return NULL;
 
   pThread->pPool = pPool;
@@ -273,7 +277,7 @@ static SThreadObj *taosGetTcpThread(SPoolObj *pPool) {
   pthread_attr_t thattr;
   pthread_attr_init(&thattr);
   pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
-  int ret = pthread_create(&(pThread->thread), &thattr, (void *) taosProcessTcpData, pThread);
+  int ret = pthread_create(&(pThread->thread), &thattr, (void *)taosProcessTcpData, pThread);
   pthread_attr_destroy(&thattr);
 
   if (ret != 0) {
@@ -290,7 +294,7 @@ static SThreadObj *taosGetTcpThread(SPoolObj *pPool) {
   return pThread;
 }
 
-static void taosStopPoolThread(SThreadObj* pThread) {
+static void taosStopPoolThread(SThreadObj *pThread) {
   pThread->stop = true;
 
   if (pThread->thread == pthread_self()) {
@@ -303,7 +307,7 @@ static void taosStopPoolThread(SThreadObj* pThread) {
 
   // signal the thread to stop, try graceful method first,
   // and use pthread_cancel when failed
-  struct epoll_event event = { .events = EPOLLIN };
+  struct epoll_event event = {.events = EPOLLIN};
   eventfd_t fd = eventfd(1, 0);
   if (fd == -1) {
     // failed to create eventfd, call pthread_cancel instead, which may result in data corruption
@@ -319,4 +323,3 @@ static void taosStopPoolThread(SThreadObj* pThread) {
   pthread_join(thread, NULL);
   taosClose(fd);
 }
-

src/sync/src/tarbitrator.c

@@ -35,7 +35,7 @@ static tsem_t   tsArbSem;
 static ttpool_h tsArbTcpPool;
 
 typedef struct {
-  char  id[TSDB_EP_LEN+24];
+  char  id[TSDB_EP_LEN + 24];
   int   nodeFd;
   void *pConn;
 } SNodeConn;
@@ -43,12 +43,12 @@ typedef struct {
 int main(int argc, char *argv[]) {
   char arbLogPath[TSDB_FILENAME_LEN + 16] = {0};
 
-  for (int i=1; i<argc; ++i) {
-    if (strcmp(argv[i], "-p")==0 && i < argc-1) {
+  for (int i = 1; i < argc; ++i) {
+    if (strcmp(argv[i], "-p") == 0 && i < argc - 1) {
       tsArbitratorPort = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-d")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) {
       debugFlag = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-g")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-g") == 0 && i < argc - 1) {
       if (strlen(argv[++i]) > TSDB_FILENAME_LEN) continue;
       tstrncpy(arbLogPath, argv[i], sizeof(arbLogPath));
     } else {
@@ -108,8 +108,7 @@ int main(int argc, char *argv[]) {
   return 0;
 }
 
-static void arbProcessIncommingConnection(int connFd, uint32_t sourceIp)
-{
+static void arbProcessIncommingConnection(int connFd, uint32_t sourceIp) {
   char ipstr[24];
   tinet_ntoa(ipstr, sourceIp);
   sDebug("peer TCP connection from ip:%s", ipstr);
@@ -121,13 +120,14 @@ static void arbProcessIncommingConnection(int connFd, uint32_t sourceIp)
     return;
   }
 
-  SNodeConn *pNode = (SNodeConn *) calloc(sizeof(SNodeConn), 1);
+  SNodeConn *pNode = (SNodeConn *)calloc(sizeof(SNodeConn), 1);
   if (pNode == NULL) {
     sError("failed to allocate memory(%s)", strerror(errno));
     taosCloseSocket(connFd);
     return;
   }
 
+  firstPkt.fqdn[sizeof(firstPkt.fqdn) - 1] = 0;
   snprintf(pNode->id, sizeof(pNode->id), "vgId:%d peer:%s:%d", firstPkt.sourceId, firstPkt.fqdn, firstPkt.port);
   if (firstPkt.syncHead.vgId) {
     sDebug("%s, vgId in head is not zero, close the connection", pNode->id);
@@ -151,10 +151,10 @@ static void arbProcessBrokenLink(void *param) {
 }
 
 static int arbProcessPeerMsg(void *param, void *buffer) {
-  SNodeConn * pNode = param;
+  SNodeConn *pNode = param;
   SSyncHead  head;
   int        bytes = 0;
-  char       *cont = (char *)buffer;
+  char *     cont = (char *)buffer;
 
   int hlen = taosReadMsg(pNode->nodeFd, &head, sizeof(head));
   if (hlen != sizeof(head)) {

src/sync/test/syncClient.c

@@ -26,14 +26,15 @@ typedef struct {
   int       numOfReqs;
   int       msgSize;
   tsem_t    rspSem;
-  tsem_t   *pOverSem; 
+  tsem_t *  pOverSem;
   pthread_t thread;
-  void     *pRpc;
+  void *    pRpc;
 } SInfo;
 
 void processResponse(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
   SInfo *pInfo = (SInfo *)pMsg->ahandle;
-  uDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen, pMsg->code);
+  uDebug("thread:%d, response is received, type:%d contLen:%d code:0x%x", pInfo->index, pMsg->msgType, pMsg->contLen,
+         pMsg->code);
 
   if (pEpSet) pInfo->epSet = *pEpSet;
   rpcFreeCont(pMsg->pCont);
@@ -44,12 +45,12 @@ void processResponse(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
 int tcount = 0;
 
 void *sendRequest(void *param) {
-  SInfo  *pInfo = (SInfo *)param;
+  SInfo * pInfo = (SInfo *)param;
   SRpcMsg rpcMsg = {0};
 
   uDebug("thread:%d, start to send request", pInfo->index);
 
-  while ( pInfo->numOfReqs == 0 || pInfo->num < pInfo->numOfReqs) {
+  while (pInfo->numOfReqs == 0 || pInfo->num < pInfo->numOfReqs) {
     pInfo->num++;
     rpcMsg.pCont = rpcMallocCont(pInfo->msgSize);
     rpcMsg.contLen = pInfo->msgSize;
@@ -57,8 +58,9 @@ void *sendRequest(void *param) {
     rpcMsg.msgType = 1;
     uDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num);
     rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg);
-    if ( pInfo->num % 20000 == 0 ) 
+    if (pInfo->num % 20000 == 0) {
       uInfo("thread:%d, %d requests have been sent", pInfo->index, pInfo->num);
+    }
     tsem_wait(&pInfo->rspSem);
   }
 
@@ -102,30 +104,30 @@ int main(int argc, char *argv[]) {
   rpcInit.spi          = 1;
   rpcInit.connType     = TAOS_CONN_CLIENT;
 
-  for (int i=1; i<argc; ++i) { 
-    if (strcmp(argv[i], "-p")==0 && i < argc-1) {
+  for (int i = 1; i < argc; ++i) {
+    if (strcmp(argv[i], "-p") == 0 && i < argc - 1) {
       epSet.port[0] = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-i") ==0 && i < argc-1) {
-      strcpy(epSet.fqdn[0], argv[++i]); 
-    } else if (strcmp(argv[i], "-t")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-i") == 0 && i < argc - 1) {
+      tstrncpy(epSet.fqdn[0], argv[++i], TSDB_FQDN_LEN);
+    } else if (strcmp(argv[i], "-t") == 0 && i < argc - 1) {
       rpcInit.numOfThreads = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-m")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-m") == 0 && i < argc - 1) {
       msgSize = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-s")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-s") == 0 && i < argc - 1) {
       rpcInit.sessions = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-n")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-n") == 0 && i < argc - 1) {
       numOfReqs = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-a")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-a") == 0 && i < argc - 1) {
       appThreads = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-o")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-o") == 0 && i < argc - 1) {
       tsCompressMsgSize = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-u")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-u") == 0 && i < argc - 1) {
       rpcInit.user = argv[++i];
-    } else if (strcmp(argv[i], "-k")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-k") == 0 && i < argc - 1) {
       rpcInit.secret = argv[++i];
-    } else if (strcmp(argv[i], "-spi")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-spi") == 0 && i < argc - 1) {
       rpcInit.spi = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-d")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) {
       rpcDebugFlag = atoi(argv[++i]);
     } else {
       printf("\nusage: %s [options] \n", argv[0]);
@@ -157,14 +159,14 @@ int main(int argc, char *argv[]) {
   uInfo("client is initialized");
 
   gettimeofday(&systemTime, NULL);
-  startTime = systemTime.tv_sec*1000000 + systemTime.tv_usec;
+  startTime = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
 
-  SInfo *pInfo = (SInfo *)calloc(1, sizeof(SInfo)*appThreads);
+  SInfo *pInfo = (SInfo *)calloc(1, sizeof(SInfo) * appThreads);
 
   pthread_attr_init(&thattr);
   pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
 
-  for (int i=0; i<appThreads; ++i) {
+  for (int i = 0; i < appThreads; ++i) {
     pInfo->index = i;
     pInfo->epSet = epSet;
     pInfo->numOfReqs = numOfReqs;
@@ -177,18 +179,16 @@ int main(int argc, char *argv[]) {
 
   do {
     usleep(1);
-  } while ( tcount < appThreads);
+  } while (tcount < appThreads);
 
   gettimeofday(&systemTime, NULL);
-  endTime = systemTime.tv_sec*1000000 + systemTime.tv_usec;  
-  float usedTime = (endTime - startTime)/1000.0;  // mseconds
+  endTime = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
+  float usedTime = (endTime - startTime) / 1000.0;  // mseconds
 
-  uInfo("it takes %.3f mseconds to send %d requests to server", usedTime, numOfReqs*appThreads);
-  uInfo("Performance: %.3f requests per second, msgSize:%d bytes", 1000.0*numOfReqs*appThreads/usedTime, msgSize);
+  uInfo("it takes %.3f mseconds to send %d requests to server", usedTime, numOfReqs * appThreads);
+  uInfo("Performance: %.3f requests per second, msgSize:%d bytes", 1000.0 * numOfReqs * appThreads / usedTime, msgSize);
 
   taosCloseLog();
 
   return 0;
 }
-
-

src/sync/test/syncServer.c

@@ -27,23 +27,22 @@
 int       msgSize = 128;
 int       commit = 0;
 int       dataFd = -1;
-void *qhandle = NULL;
+void *    qhandle = NULL;
 int       walNum = 0;
 uint64_t  tversion = 0;
-void *syncHandle;
+void *    syncHandle;
 int       role;
 int       nodeId;
 char      path[256];
-int  numOfWrites ;
+int       numOfWrites;
 SSyncInfo syncInfo;
 SSyncCfg *pCfg;
 
-int writeIntoWal(SWalHead *pHead)
-{ 
+int writeIntoWal(SWalHead *pHead) {
   if (dataFd < 0) {
     char walName[280];
     snprintf(walName, sizeof(walName), "%s/wal/wal.%d", path, walNum);
-    remove(walName);
+    (void)remove(walName);
     dataFd = open(walName, O_CREAT | O_WRONLY, S_IRWXU | S_IRWXG | S_IRWXO);
     if (dataFd < 0) {
       uInfo("failed to open wal file:%s(%s)", walName, strerror(errno));
@@ -71,16 +70,15 @@ int writeIntoWal(SWalHead *pHead)
   return 0;
 }
 
-void confirmForward(void *ahandle, void *mhandle, int32_t code)
-{
-  SRpcMsg  *pMsg = (SRpcMsg *)mhandle;
+void confirmForward(void *ahandle, void *mhandle, int32_t code) {
+  SRpcMsg * pMsg = (SRpcMsg *)mhandle;
   SWalHead *pHead = (SWalHead *)(((char *)pMsg->pCont) - sizeof(SWalHead));
 
   uDebug("ver:%" PRIu64 ", confirm is received", pHead->version);
 
   rpcFreeCont(pMsg->pCont);
 
-  SRpcMsg    rpcMsg;
+  SRpcMsg rpcMsg = {0};
   rpcMsg.pCont = rpcMallocCont(msgSize);
   rpcMsg.contLen = msgSize;
   rpcMsg.handle = pMsg->handle;
@@ -91,14 +89,13 @@ void confirmForward(void *ahandle, void *mhandle, int32_t code)
 }
 
 int processRpcMsg(void *item) {
-  SRpcMsg   *pMsg = (SRpcMsg *)item;
+  SRpcMsg * pMsg = (SRpcMsg *)item;
   SWalHead *pHead = (SWalHead *)(((char *)pMsg->pCont) - sizeof(SWalHead));
   int       code = -1;
 
   if (role != TAOS_SYNC_ROLE_MASTER) {
     uError("not master, write failed, role:%s", syncRole[role]);
   } else {
-
     pHead->version = ++tversion;
     pHead->msgType = pMsg->msgType;
     pHead->len = pMsg->contLen;
@@ -111,10 +108,10 @@ int processRpcMsg(void *item) {
   }
 
   if (pCfg->quorum <= 1) {
-    taosFreeQitem(item); 
     rpcFreeCont(pMsg->pCont);
+    taosFreeQitem(item);
 
-    SRpcMsg    rpcMsg;
+    SRpcMsg rpcMsg = {0};
     rpcMsg.pCont = rpcMallocCont(msgSize);
     rpcMsg.contLen = msgSize;
     rpcMsg.handle = pMsg->handle;
@@ -126,7 +123,6 @@ int processRpcMsg(void *item) {
 }
 
 int processFwdMsg(void *item) {
-
   SWalHead *pHead = (SWalHead *)item;
 
   if (pHead->version <= tversion) {
@@ -142,11 +138,11 @@ int processFwdMsg(void *item) {
 
   // write into cache
 
-/*
+  /*
     if (pHead->handle) {
       syncSendFwdAck(syncHandle, pHead->handle, 0);
     }
-*/
+  */
 
   taosFreeQitem(item);
 
@@ -154,7 +150,6 @@ int processFwdMsg(void *item) {
 }
 
 int processWalMsg(void *item) {
-
   SWalHead *pHead = (SWalHead *)item;
 
   if (pHead->version <= tversion) {
@@ -168,11 +163,11 @@ int processWalMsg(void *item) {
 
   // write into cache
 
-/*
+  /*
     if (pHead->handle) {
       syncSendFwdAck(syncHandle, pHead->handle, 0);
     }
-*/
+  */
 
   taosFreeQitem(item);
 
@@ -197,7 +192,6 @@ void *processWriteQueue(void *param) {
     } else if (type == TAOS_QTYPE_FWD) {
       processFwdMsg(item);
     }
-
   }
 
   return NULL;
@@ -224,7 +218,6 @@ int retrieveAuthInfo(char *meterId, char *spi, char *encrypt, char *secret, char
 }
 
 void processRequestMsg(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
-  
   SRpcMsg *pTemp;
 
   pTemp = taosAllocateQitem(sizeof(SRpcMsg));
@@ -234,8 +227,7 @@ void processRequestMsg(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
   taosWriteQitem(qhandle, TAOS_QTYPE_RPC, pTemp);
 }
 
-uint32_t getFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex, int64_t *size, uint64_t *fversion) 
-{
+uint32_t getFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex, int64_t *size, uint64_t *fversion) {
   uint32_t    magic;
   struct stat fstat;
   char        aname[280];
@@ -254,7 +246,7 @@ uint32_t getFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex
   }
 
   uInfo("get file info:%s", aname);
-  if ( stat(aname, &fstat) < 0 ) return 0; 
+  if (stat(aname, &fstat) < 0) return 0;
 
   *size = fstat.st_size;
   magic = fstat.st_size;
@@ -263,23 +255,21 @@ uint32_t getFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex
 }
 
 int getWalInfo(void *ahandle, char *name, uint32_t *index) {
-
   struct stat fstat;
   char        aname[280];
 
   name[0] = 0;
-  if (*index + 1> walNum) return 0;
+  if (*index + 1 > walNum) return 0;
 
   snprintf(aname, sizeof(aname), "%s/wal/wal.%d", path, *index);
   sprintf(name, "wal/wal.%d", *index);
   uInfo("get wal info:%s", aname);
 
-  if ( stat(aname, &fstat) < 0 ) return -1;
+  if (stat(aname, &fstat) < 0) return -1;
 
-  if (*index >= walNum-1) return 0;  // no more
+  if (*index >= walNum - 1) return 0;  // no more
 
   return 1;
-
 }
 
 int writeToCache(void *ahandle, void *data, int type) {
@@ -295,19 +285,14 @@ int writeToCache(void *ahandle, void *data, int type) {
   return 0;
 }
 
-void confirmFwd(void *ahandle, int64_t version) {
-
-  return;
-}
+void confirmFwd(void *ahandle, int64_t version) { return; }
 
 void notifyRole(void *ahandle, int8_t r) {
   role = r;
   printf("current role:%s\n", syncRole[role]);
 }
 
-
 void initSync() {
-
   pCfg->replica = 1;
   pCfg->quorum = 1;
   syncInfo.vgId = 1;
@@ -339,17 +324,15 @@ void initSync() {
   taosGetFqdn(pCfg->nodeInfo[4].nodeFqdn);
 }
 
-void doSync()
-{
-  for (int i=0; i<5; ++i) {
-    if (tsSyncPort == pCfg->nodeInfo[i].nodePort) 
-      nodeId = pCfg->nodeInfo[i].nodeId;
+void doSync() {
+  for (int i = 0; i < 5; ++i) {
+    if (tsSyncPort == pCfg->nodeInfo[i].nodePort) nodeId = pCfg->nodeInfo[i].nodeId;
   }
 
   snprintf(path, sizeof(path), "/root/test/d%d", nodeId);
-  strcpy(syncInfo.path, path);
+  tstrncpy(syncInfo.path, path, sizeof(syncInfo.path));
 
-  if ( syncHandle == NULL) {
+  if (syncHandle == NULL) {
     syncHandle = syncStart(&syncInfo);
   } else {
     if (syncReconfig(syncHandle, pCfg) < 0) syncHandle = NULL;
@@ -371,29 +354,29 @@ int main(int argc, char *argv[]) {
   rpcInit.numOfThreads = 1;
   rpcInit.cfp = processRequestMsg;
   rpcInit.sessions = 1000;
-  rpcInit.idleTime     = tsShellActivityTimer*1500; 
+  rpcInit.idleTime = tsShellActivityTimer * 1500;
   rpcInit.afp = retrieveAuthInfo;
 
-  for (int i=1; i<argc; ++i) {
-    if (strcmp(argv[i], "-p")==0 && i < argc-1) {
+  for (int i = 1; i < argc; ++i) {
+    if (strcmp(argv[i], "-p") == 0 && i < argc - 1) {
       rpcInit.localPort = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-t")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-t") == 0 && i < argc - 1) {
       rpcInit.numOfThreads = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-m")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-m") == 0 && i < argc - 1) {
       msgSize = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-s")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-s") == 0 && i < argc - 1) {
       rpcInit.sessions = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-o")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-o") == 0 && i < argc - 1) {
       tsCompressMsgSize = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-w")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-w") == 0 && i < argc - 1) {
       commit = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-v")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-v") == 0 && i < argc - 1) {
       syncInfo.version = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-r")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-r") == 0 && i < argc - 1) {
       pCfg->replica = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-q")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-q") == 0 && i < argc - 1) {
       pCfg->quorum = atoi(argv[++i]);
-    } else if (strcmp(argv[i], "-d")==0 && i < argc-1) {
+    } else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) {
       rpcDebugFlag = atoi(argv[++i]);
     } else {
       printf("\nusage: %s [options] \n", argv[0]);
@@ -414,7 +397,7 @@ int main(int argc, char *argv[]) {
 
   uDebugFlag = rpcDebugFlag;
   dDebugFlag = rpcDebugFlag;
-  //tmrDebugFlag = rpcDebugFlag; 
+  // tmrDebugFlag = rpcDebugFlag;
   tsAsyncLog = 0;
   taosInitLog("server.log", 1000000, 10);
 
@@ -443,35 +426,39 @@ int main(int argc, char *argv[]) {
   SNodesRole nroles;
 
   while (1) {
-    char c = getchar();
+    int c = getchar();
 
-    switch(c) {
+    switch (c) {
       case '1':
-        pCfg->replica = 1; doSync();
+        pCfg->replica = 1;
+        doSync();
         break;
       case '2':
-        pCfg->replica = 2; doSync();
+        pCfg->replica = 2;
+        doSync();
         break;
       case '3':
-        pCfg->replica = 3; doSync();
+        pCfg->replica = 3;
+        doSync();
         break;
       case '4':
-        pCfg->replica = 4; doSync();
+        pCfg->replica = 4;
+        doSync();
         break;
       case '5':
-        pCfg->replica = 5; doSync();
+        pCfg->replica = 5;
+        doSync();
         break;
       case 's':
         syncGetNodesRole(syncHandle, &nroles);
-        for (int i=0; i<pCfg->replica; ++i) 
+        for (int i = 0; i < pCfg->replica; ++i)
           printf("=== nodeId:%d role:%s\n", nroles.nodeId[i], syncRole[nroles.role[i]]);
         break;
       default:
         break;
     }
 
-    if (c=='q') break;
-
+    if (c == 'q') break;
   }
 
   syncStop(syncHandle);
@@ -483,5 +470,3 @@ int main(int argc, char *argv[]) {
 
   return 0;
 }
-
-