TD-1520

src/cq/src/cqMain.c

@@ -30,10 +30,12 @@
 #include "tlog.h"
 #include "twal.h"
 
-#define cError(...) { if (cqDebugFlag & DEBUG_ERROR) { taosPrintLog("ERROR CQ  ", cqDebugFlag, __VA_ARGS__); }}
-#define cWarn(...)  { if (cqDebugFlag & DEBUG_WARN)  { taosPrintLog("WARN CQ  ", cqDebugFlag, __VA_ARGS__); }}
+#define cFatal(...) { if (cqDebugFlag & DEBUG_FATAL) { taosPrintLog("CQ  FATAL ", 255, __VA_ARGS__); }}
+#define cError(...) { if (cqDebugFlag & DEBUG_ERROR) { taosPrintLog("CQ  ERROR ", 255, __VA_ARGS__); }}
+#define cWarn(...)  { if (cqDebugFlag & DEBUG_WARN)  { taosPrintLog("CQ  WARN ", 255, __VA_ARGS__); }}
+#define cInfo(...)  { if (cqDebugFlag & DEBUG_INFO)  { taosPrintLog("CQ  ", 255, __VA_ARGS__); }}
+#define cDebug(...) { if (cqDebugFlag & DEBUG_DEBUG) { taosPrintLog("CQ  ", cqDebugFlag, __VA_ARGS__); }}
 #define cTrace(...) { if (cqDebugFlag & DEBUG_TRACE) { taosPrintLog("CQ  ", cqDebugFlag, __VA_ARGS__); }}
-#define cPrint(...) { taosPrintLog("CQ  ", 255, __VA_ARGS__); }
 
 typedef struct {
   int      vgId;
@@ -94,7 +96,7 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
 
   pthread_mutex_init(&pContext->mutex, NULL);
 
-  cTrace("vgId:%d, CQ is opened", pContext->vgId);
+  cInfo("vgId:%d, CQ is opened", pContext->vgId);
 
   return pContext;
 }
@@ -125,7 +127,7 @@ void cqClose(void *handle) {
   taosTmrCleanUp(pContext->tmrCtrl);
   pContext->tmrCtrl = NULL;
 
-  cTrace("vgId:%d, CQ is closed", pContext->vgId);
+  cInfo("vgId:%d, CQ is closed", pContext->vgId);
   free(pContext);
 }
 
@@ -133,7 +135,7 @@ void cqStart(void *handle) {
   SCqContext *pContext = handle;
   if (pContext->dbConn || pContext->master) return;
 
-  cTrace("vgId:%d, start all CQs", pContext->vgId);
+  cInfo("vgId:%d, start all CQs", pContext->vgId);
   pthread_mutex_lock(&pContext->mutex);
 
   pContext->master = 1;
@@ -149,7 +151,7 @@ void cqStart(void *handle) {
 
 void cqStop(void *handle) {
   SCqContext *pContext = handle;
-  cTrace("vgId:%d, stop all CQs", pContext->vgId);
+  cInfo("vgId:%d, stop all CQs", pContext->vgId);
   if (pContext->dbConn == NULL || pContext->master == 0) return;
 
   pthread_mutex_lock(&pContext->mutex);
@@ -160,7 +162,7 @@ void cqStop(void *handle) {
     if (pObj->pStream) {
       taos_close_stream(pObj->pStream);
       pObj->pStream = NULL;
-      cTrace("vgId:%d, id:%d CQ:%s is closed", pContext->vgId, pObj->tid, pObj->sqlStr);
+      cInfo("vgId:%d, id:%d CQ:%s is closed", pContext->vgId, pObj->tid, pObj->sqlStr);
     } else {
       taosTmrStop(pObj->tmrId);
       pObj->tmrId = 0;
@@ -188,7 +190,7 @@ void *cqCreate(void *handle, uint64_t uid, int tid, char *sqlStr, STSchema *pSch
   pObj->pSchema = tdDupSchema(pSchema);
   pObj->rowSize = schemaTLen(pSchema);
 
-  cTrace("vgId:%d, id:%d CQ:%s is created", pContext->vgId, pObj->tid, pObj->sqlStr);
+  cInfo("vgId:%d, id:%d CQ:%s is created", pContext->vgId, pObj->tid, pObj->sqlStr);
 
   pthread_mutex_lock(&pContext->mutex);
 
@@ -228,7 +230,7 @@ void cqDrop(void *handle) {
     pObj->tmrId = 0;
   }
 
-  cTrace("vgId:%d, id:%d CQ:%s is dropped", pContext->vgId, pObj->tid, pObj->sqlStr); 
+  cInfo("vgId:%d, id:%d CQ:%s is dropped", pContext->vgId, pObj->tid, pObj->sqlStr); 
   tdFreeSchema(pObj->pSchema);
   free(pObj->sqlStr);
   free(pObj);
@@ -262,7 +264,7 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
   pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, 0, pObj, NULL);
   if (pObj->pStream) {
     pContext->num++;
-    cTrace("vgId:%d, id:%d CQ:%s is openned", pContext->vgId, pObj->tid, pObj->sqlStr);
+    cInfo("vgId:%d, id:%d CQ:%s is openned", pContext->vgId, pObj->tid, pObj->sqlStr);
   } else {
     cError("vgId:%d, id:%d CQ:%s, failed to open", pContext->vgId, pObj->tid, pObj->sqlStr);
   }
@@ -278,7 +280,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
   STSchema   *pSchema = pObj->pSchema;
   if (pObj->pStream == NULL) return;
 
-  cTrace("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->tid, pObj->sqlStr);
+  cDebug("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->tid, pObj->sqlStr);
 
   int size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + TD_DATA_ROW_HEAD_SIZE + pObj->rowSize;
   char *buffer = calloc(size, 1);

src/sync/src/syncMain.c

@@ -513,7 +513,7 @@ static SSyncPeer *syncAddPeer(SSyncNode *pNode, const SNodeInfo *pInfo) {
   if (pPeer->nodeId == 0 || (ret > 0) || (ret == 0 && pPeer->port > tsSyncPort)) {
     sDebug("%s, start to check peer connection", pPeer->id);
     int32_t checkMs = 100 + (pNode->vgId * 10) % 100;
-    if (pNode->vgId) checkMs = tsStatusInterval * 2000 + 100;
+    if (pNode->vgId > 1) checkMs = tsStatusInterval * 2000 + checkMs;
     taosTmrReset(syncCheckPeerConnection, checkMs, pPeer, syncTmrCtrl, &pPeer->timer);
   }