add timeout to auth

src/dnode/src/dnodePeer.c

@@ -20,13 +20,13 @@
  */
 
 #define _DEFAULT_SOURCE
-#include "os.h"
-#include "mnode.h"
-#include "dnodeVMgmt.h"
-#include "dnodeVWrite.h"
-#include "dnodeMPeer.h"
 #include "dnodeMInfos.h"
+#include "dnodeMPeer.h"
 #include "dnodeStep.h"
+#include "dnodeVMgmt.h"
+#include "dnodeVWrite.h"
+#include "mnode.h"
+#include "os.h"
 
 static void (*dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
 static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcEpSet *);
@@ -85,11 +85,7 @@ void dnodeCleanupServer() {
 }
 
 static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
-  SRpcMsg rspMsg = {
-    .handle  = pMsg->handle,
-    .pCont   = NULL,
-    .contLen = 0
-  };
+  SRpcMsg rspMsg = {.handle = pMsg->handle, .pCont = NULL, .contLen = 0};
 
   if (pMsg->pCont == NULL) return;
   if (pMsg->msgType == TSDB_MSG_TYPE_NETWORK_TEST) {
@@ -174,13 +170,9 @@ static void dnodeProcessRspFromDnode(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
   rpcFreeCont(pMsg->pCont);
 }
 
-void dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg)) {
-  dnodeProcessRspMsgFp[msgType] = fp;
-}
+void dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg)) { dnodeProcessRspMsgFp[msgType] = fp; }
 
-void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg) {
-  rpcSendRequest(tsClientRpc, epSet, rpcMsg, NULL);
-}
+void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg) { rpcSendRequest(tsClientRpc, epSet, rpcMsg, NULL); }
 
 void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) {
   SRpcEpSet epSet = {0};
@@ -189,6 +181,13 @@ void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) {
   assert(tsClientRpc != 0);
   rpcSendRecv(tsClientRpc, &epSet, rpcMsg, rpcRsp);
 }
+void dnodeSendMsgToMnodeRecvWithTimeout(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) {
+  SRpcEpSet epSet = {0};
+  dnodeGetEpSetForPeer(&epSet);
+
+  assert(tsClientRpc != 0);
+  rpcSendRecvWithTimeout(tsClientRpc, &epSet, rpcMsg, rpcRsp);
+}
 
 void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet) {
   rpcSendRecv(tsClientRpc, epSet, rpcMsg, rpcRsp);

src/dnode/src/dnodeShell.c

@@ -14,20 +14,20 @@
  */
 
 #define _DEFAULT_SOURCE
-#include "os.h"
-#include "http.h"
-#include "mnode.h"
-#include "dnodeVRead.h"
-#include "dnodeVWrite.h"
+#include "dnodeShell.h"
 #include "dnodeMRead.h"
 #include "dnodeMWrite.h"
-#include "dnodeShell.h"
 #include "dnodeStep.h"
+#include "dnodeVRead.h"
+#include "dnodeVWrite.h"
+#include "http.h"
+#include "mnode.h"
+#include "os.h"
 
 static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
 static void    dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *);
 static int     dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
-static void  * tsShellRpc = NULL;
+static void   *tsShellRpc = NULL;
 static int64_t tsQueryReqNum = 0;
 static int64_t tsSubmitReqNum = 0;
 
@@ -44,7 +44,7 @@ int32_t dnodeInitShell() {
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_USER] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_USER] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_USER] = dnodeDispatchToMWriteQueue;
-  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DNODE]= dnodeDispatchToMWriteQueue;
+  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DNODE] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DNODE] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DB] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_TP] = dnodeDispatchToMWriteQueue;
@@ -55,22 +55,22 @@ int32_t dnodeInitShell() {
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_FUNCTION] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_DB] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_TP] = dnodeDispatchToMWriteQueue;
-  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_TABLE]= dnodeDispatchToMWriteQueue;
+  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_TABLE] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_TABLE] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_TABLE] = dnodeDispatchToMWriteQueue;
-  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_STREAM]= dnodeDispatchToMWriteQueue;
+  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_STREAM] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_QUERY] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_STREAM] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_CONN] = dnodeDispatchToMWriteQueue;
-  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONFIG_DNODE]= dnodeDispatchToMWriteQueue;
-  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_COMPACT_VNODE]= dnodeDispatchToMWriteQueue;
+  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONFIG_DNODE] = dnodeDispatchToMWriteQueue;
+  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_COMPACT_VNODE] = dnodeDispatchToMWriteQueue;
 
   // the following message shall be treated as mnode query
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_HEARTBEAT] = dnodeDispatchToMReadQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONNECT] = dnodeDispatchToMReadQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_USE_DB] = dnodeDispatchToMReadQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLE_META] = dnodeDispatchToMReadQueue;
-  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_STABLE_VGROUP]= dnodeDispatchToMReadQueue;
+  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_STABLE_VGROUP] = dnodeDispatchToMReadQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLES_META] = dnodeDispatchToMReadQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_SHOW] = dnodeDispatchToMReadQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_RETRIEVE] = dnodeDispatchToMReadQueue;
@@ -113,11 +113,7 @@ void dnodeCleanupShell() {
 }
 
 static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
-  SRpcMsg rpcMsg = {
-    .handle  = pMsg->handle,
-    .pCont   = NULL,
-    .contLen = 0
-  };
+  SRpcMsg rpcMsg = {.handle = pMsg->handle, .pCont = NULL, .contLen = 0};
 
   if (pMsg->pCont == NULL) return;
 
@@ -148,9 +144,10 @@ static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
     atomic_fetch_add_64(&tsQueryReqNum, 1);
   } else if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) {
     atomic_fetch_add_64(&tsSubmitReqNum, 1);
-  } else {}
+  } else {
+  }
 
-  if ( dnodeProcessShellMsgFp[pMsg->msgType] ) {
+  if (dnodeProcessShellMsgFp[pMsg->msgType]) {
     (*dnodeProcessShellMsgFp[pMsg->msgType])(pMsg);
   } else {
     dError("RPC %p, shell msg:%s is not processed", pMsg->handle, taosMsg[pMsg->msgType]);
@@ -191,10 +188,10 @@ static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char
 
   dDebug("user:%s, send auth msg to mnodes", user);
   SRpcMsg rpcRsp = {0};
-  dnodeSendMsgToMnodeRecv(&rpcMsg, &rpcRsp);
+  dnodeSendMsgToMnodeRecvWithTimeout(&rpcMsg, &rpcRsp);
 
   if (rpcRsp.code != 0) {
-    dError("user:%s, auth msg received from mnodes, error:%s", user, tstrerror(rpcRsp.code));
+    dError("user:%s, auth msg received from mnodes, error:%s, may be timeout", user, tstrerror(rpcRsp.code));
   } else {
     SAuthRsp *pRsp = rpcRsp.pCont;
     dDebug("user:%s, auth msg received from mnodes", user);
@@ -240,7 +237,8 @@ void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid) {
     int16_t            numOfTags = htons(pTable->numOfTags);
     int32_t            tableId = htonl(pTable->tid);
     uint64_t           uid = htobe64(pTable->uid);
-    dInfo("table:%s, numOfColumns:%d numOfTags:%d tid:%d uid:%" PRIu64, pTable->tableFname, numOfColumns, numOfTags, tableId, uid);
+    dInfo("table:%s, numOfColumns:%d numOfTags:%d tid:%d uid:%" PRIu64, pTable->tableFname, numOfColumns, numOfTags,
+          tableId, uid);
 
     return rpcRsp.pCont;
   }

src/inc/dnode.h

@@ -20,8 +20,8 @@
 extern "C" {
 #endif
 
-#include "trpc.h"
 #include "taosmsg.h"
+#include "trpc.h"
 
 #define MAX_HTTP_STATUS_CODE_NUM 63
 typedef struct {
@@ -50,6 +50,7 @@ int32_t dnodeStartMnode(SMInfos *pMinfos);
 void  dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg));
 void  dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg);
 void  dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp);
+void  dnodeSendMsgToMnodeRecvWithTimeout(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp);
 void  dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet);
 void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid);
 

src/inc/trpc.h

@@ -90,13 +90,14 @@ void  rpcSendResponse(const SRpcMsg *pMsg);
 void    rpcSendRedirectRsp(void *pConn, const SRpcEpSet *pEpSet);
 int     rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo);
 void    rpcSendRecv(void *shandle, SRpcEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp);
+void    rpcSendRecvWithTimeout(void *shandle, SRpcEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp);
 int     rpcReportProgress(void *pConn, char *pCont, int contLen);
 void    rpcCancelRequest(int64_t rid);
-int32_t rpcUnusedSession(void * rpcInfo, bool bLock);
+int32_t rpcUnusedSession(void *rpcInfo, bool bLock);
 // send rpc Refid connection probe alive message
-bool rpcSendProbe(int64_t rpcRid, void* pPrevContext, bool *pReqOver);
+bool rpcSendProbe(int64_t rpcRid, void *pPrevContext, bool *pReqOver);
 // after sql request send , save conn info
-bool rpcSaveSendInfo(int64_t rpcRid, void** ppContext);
+bool rpcSaveSendInfo(int64_t rpcRid, void **ppContext);
 
 #ifdef __cplusplus
 }

src/os/inc/osSemaphore.h

@@ -20,42 +20,46 @@
 extern "C" {
 #endif
 
-#if defined (_TD_DARWIN_64)
-  typedef struct tsem_s *tsem_t;
-  int tsem_init(tsem_t *sem, int pshared, unsigned int value);
-  int tsem_wait(tsem_t *sem);
-  int tsem_post(tsem_t *sem);
-  int tsem_destroy(tsem_t *sem);
+#include <semaphore.h>
+#if defined(_TD_DARWIN_64)
+typedef struct tsem_s *tsem_t;
+int                    tsem_init(tsem_t *sem, int pshared, unsigned int value);
+int                    tsem_wait(tsem_t *sem);
+int                    tsem_post(tsem_t *sem);
+int                    tsem_timewait(tsem_t *sim, int64_t milis);
+int                    tsem_destroy(tsem_t *sem);
+
 #else
-  #define tsem_t sem_t
-  #define tsem_init sem_init
-  int tsem_wait(tsem_t* sem);
-  #define tsem_post sem_post
-  #define tsem_destroy sem_destroy
+#define tsem_t sem_t
+#define tsem_init sem_init
+int tsem_wait(tsem_t *sem);
+int tsem_timewait(tsem_t *sim, int64_t milis);
+#define tsem_post sem_post
+#define tsem_destroy sem_destroy
 #endif
 
-#if defined (_TD_DARWIN_64)
-  #define pthread_rwlock_t pthread_mutex_t
-  #define pthread_rwlock_init(lock, NULL) pthread_mutex_init(lock, NULL)
-  #define pthread_rwlock_destroy(lock) pthread_mutex_destroy(lock)
-  #define pthread_rwlock_wrlock(lock) pthread_mutex_lock(lock)
-  #define pthread_rwlock_rdlock(lock) pthread_mutex_lock(lock)
-  #define pthread_rwlock_unlock(lock) pthread_mutex_unlock(lock)
+#if defined(_TD_DARWIN_64)
+#define pthread_rwlock_t pthread_mutex_t
+#define pthread_rwlock_init(lock, NULL) pthread_mutex_init(lock, NULL)
+#define pthread_rwlock_destroy(lock) pthread_mutex_destroy(lock)
+#define pthread_rwlock_wrlock(lock) pthread_mutex_lock(lock)
+#define pthread_rwlock_rdlock(lock) pthread_mutex_lock(lock)
+#define pthread_rwlock_unlock(lock) pthread_mutex_unlock(lock)
 
-  #define pthread_spinlock_t pthread_mutex_t
-  #define pthread_spin_init(lock, NULL) pthread_mutex_init(lock, NULL)
-  #define pthread_spin_destroy(lock) pthread_mutex_destroy(lock)
-  #define pthread_spin_lock(lock) pthread_mutex_lock(lock)
-  #define pthread_spin_unlock(lock) pthread_mutex_unlock(lock)
+#define pthread_spinlock_t pthread_mutex_t
+#define pthread_spin_init(lock, NULL) pthread_mutex_init(lock, NULL)
+#define pthread_spin_destroy(lock) pthread_mutex_destroy(lock)
+#define pthread_spin_lock(lock) pthread_mutex_lock(lock)
+#define pthread_spin_unlock(lock) pthread_mutex_unlock(lock)
 #endif
 
 bool    taosCheckPthreadValid(pthread_t thread);
 int64_t taosGetSelfPthreadId();
 int64_t taosGetPthreadId(pthread_t thread);
-void    taosResetPthread(pthread_t* thread);
+void    taosResetPthread(pthread_t *thread);
 bool    taosComparePthread(pthread_t first, pthread_t second);
 int32_t taosGetPId();
-int32_t taosGetCurrentAPPName(char* name, int32_t* len);
+int32_t taosGetCurrentAPPName(char *name, int32_t *len);
 
 #ifdef __cplusplus
 }

src/rpc/src/rpcMain.c

@@ -561,6 +561,29 @@ void rpcSendRecv(void *shandle, SRpcEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp)
   return;
 }
 
+void rpcSendRecvWithTimeout(void *shandle, SRpcEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
+  SRpcReqContext *pContext;
+  pContext = (SRpcReqContext *)((char *)pMsg->pCont - sizeof(SRpcHead) - sizeof(SRpcReqContext));
+
+  memset(pRsp, 0, sizeof(SRpcMsg));
+
+  tsem_t sem;
+  tsem_init(&sem, 0, 0);
+  pContext->pSem = &sem;
+  pContext->pRsp = pRsp;
+  pContext->pSet = pEpSet;
+
+  int64_t rid = 0;
+  rpcSendRequest(shandle, pEpSet, pMsg, &rid);
+
+  tsem_timewait(&sem, 3 * 1000);
+  rpcCancelRequest(rid);
+  tsem_destroy(&sem);
+
+  pRsp->code = -1;
+  return;
+}
+
 // this API is used by server app to keep an APP context in case connection is broken
 int rpcReportProgress(void *handle, char *pCont, int contLen) {
   SRpcConn *pConn = (SRpcConn *)handle;