Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
76acde1e
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
76acde1e
编写于
4月 24, 2022
作者:
S
Shengliang Guan
提交者:
GitHub
4月 24, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #11771 from taosdata/fix/hb
fix: heartbeat problem
上级
d66d638c
ee929937
变更
22
展开全部
显示空白变更内容
内联
并排
Showing
22 changed file
with
1562 addition
and
1214 deletion
+1562
-1214
include/libs/transport/trpc.h
include/libs/transport/trpc.h
+12
-11
source/dnode/mgmt/implement/src/dmTransport.c
source/dnode/mgmt/implement/src/dmTransport.c
+15
-5
source/dnode/mgmt/implement/src/dmWorker.c
source/dnode/mgmt/implement/src/dmWorker.c
+2
-2
source/dnode/mgmt/interface/src/dmInt.c
source/dnode/mgmt/interface/src/dmInt.c
+1
-1
source/dnode/mgmt/mgmt_bnode/src/bmWorker.c
source/dnode/mgmt/mgmt_bnode/src/bmWorker.c
+5
-3
source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
+1
-0
source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
+1
-0
source/dnode/mgmt/mgmt_snode/src/smWorker.c
source/dnode/mgmt/mgmt_snode/src/smWorker.c
+5
-4
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
+13
-10
source/dnode/mnode/impl/inc/mndDef.h
source/dnode/mnode/impl/inc/mndDef.h
+1
-0
source/dnode/mnode/impl/src/mndTrans.c
source/dnode/mnode/impl/src/mndTrans.c
+15
-12
source/dnode/vnode/src/vnd/vnodeQuery.c
source/dnode/vnode/src/vnd/vnodeQuery.c
+6
-5
source/dnode/vnode/src/vnd/vnodeSync.c
source/dnode/vnode/src/vnd/vnodeSync.c
+2
-1
source/libs/executor/src/executorimpl.c
source/libs/executor/src/executorimpl.c
+619
-518
source/libs/qworker/inc/qworkerInt.h
source/libs/qworker/inc/qworkerInt.h
+155
-110
source/libs/qworker/src/qworker.c
source/libs/qworker/src/qworker.c
+309
-338
source/libs/qworker/src/qworkerMsg.c
source/libs/qworker/src/qworkerMsg.c
+151
-139
source/libs/scheduler/src/scheduler.c
source/libs/scheduler/src/scheduler.c
+13
-13
source/libs/sync/src/syncIO.c
source/libs/sync/src/syncIO.c
+1
-1
source/libs/transport/inc/transComm.h
source/libs/transport/inc/transComm.h
+4
-0
source/libs/transport/src/transComm.c
source/libs/transport/src/transComm.c
+8
-1
source/libs/transport/src/transSrv.c
source/libs/transport/src/transSrv.c
+223
-40
未找到文件。
include/libs/transport/trpc.h
浏览文件 @
76acde1e
...
...
@@ -38,11 +38,12 @@ typedef struct SRpcConnInfo {
typedef
struct
SRpcMsg
{
tmsg_t
msgType
;
void
*
pCont
;
void
*
pCont
;
int
contLen
;
int32_t
code
;
void
*
handle
;
// rpc handle returned to app
void
*
ahandle
;
// app handle set by client
void
*
handle
;
// rpc handle returned to app
void
*
ahandle
;
// app handle set by client
int64_t
refId
;
// refid, used by server
int
noResp
;
// has response or not(default 0, 0: resp, 1: no resp);
int
persistHandle
;
// persist handle or not
...
...
@@ -54,8 +55,8 @@ typedef struct {
uint16_t
clientPort
;
SRpcMsg
rpcMsg
;
int32_t
rspLen
;
void
*
pRsp
;
void
*
pNode
;
void
*
pRsp
;
void
*
pNode
;
}
SNodeMsg
;
typedef
void
(
*
RpcCfp
)(
void
*
parent
,
SRpcMsg
*
,
SEpSet
*
);
...
...
@@ -64,7 +65,7 @@ typedef int (*RpcRfp)(void *parent, SRpcMsg *, SEpSet *);
typedef
struct
SRpcInit
{
uint16_t
localPort
;
// local port
char
*
label
;
// for debug purpose
char
*
label
;
// for debug purpose
int
numOfThreads
;
// number of threads to handle connections
int
sessions
;
// number of sessions allowed
int8_t
connType
;
// TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
...
...
@@ -97,23 +98,23 @@ typedef struct {
typedef
struct
{
int32_t
msgType
;
void
*
val
;
void
*
val
;
int32_t
(
*
clone
)(
void
*
src
,
void
**
dst
);
void
(
*
freeFunc
)(
const
void
*
arg
);
}
SRpcBrokenlinkVal
;
typedef
struct
{
SHashObj
*
args
;
SHashObj
*
args
;
SRpcBrokenlinkVal
brokenVal
;
}
SRpcCtx
;
int32_t
rpcInit
();
void
rpcCleanup
();
void
*
rpcOpen
(
const
SRpcInit
*
pRpc
);
void
*
rpcOpen
(
const
SRpcInit
*
pRpc
);
void
rpcClose
(
void
*
);
void
*
rpcMallocCont
(
int
contLen
);
void
*
rpcMallocCont
(
int
contLen
);
void
rpcFreeCont
(
void
*
pCont
);
void
*
rpcReallocCont
(
void
*
ptr
,
int
contLen
);
void
*
rpcReallocCont
(
void
*
ptr
,
int
contLen
);
// Because taosd supports multi-process mode
// These functions should not be used on the server side
...
...
source/dnode/mgmt/implement/src/dmTransport.c
浏览文件 @
76acde1e
...
...
@@ -59,6 +59,10 @@ static inline int32_t dmBuildMsg(SNodeMsg *pMsg, SRpcMsg *pRpc) {
pMsg
->
clientIp
=
connInfo
.
clientIp
;
pMsg
->
clientPort
=
connInfo
.
clientPort
;
memcpy
(
&
pMsg
->
rpcMsg
,
pRpc
,
sizeof
(
SRpcMsg
));
if
((
pRpc
->
msgType
&
1u
))
{
assert
(
pRpc
->
refId
!=
0
);
}
// assert(pRpc->handle != NULL && pRpc->refId != 0 && pMsg->rpcMsg.refId != 0);
return
0
;
}
...
...
@@ -107,7 +111,7 @@ _OVER:
}
}
SRpcMsg
rsp
=
{.
handle
=
pRpc
->
handle
,
.
ahandle
=
pRpc
->
ahandle
,
.
code
=
code
};
SRpcMsg
rsp
=
{.
handle
=
pRpc
->
handle
,
.
ahandle
=
pRpc
->
ahandle
,
.
code
=
code
,
.
refId
=
pRpc
->
refId
};
tmsgSendRsp
(
&
rsp
);
}
dTrace
(
"msg:%p, is freed"
,
pMsg
);
...
...
@@ -134,7 +138,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
if
(
pDnode
->
status
!=
DND_STAT_RUNNING
)
{
dError
(
"msg:%s ignored since dnode not running, handle:%p app:%p"
,
TMSG_INFO
(
msgType
),
pMsg
->
handle
,
pMsg
->
ahandle
);
if
(
isReq
)
{
SRpcMsg
rspMsg
=
{.
handle
=
pMsg
->
handle
,
.
code
=
TSDB_CODE_APP_NOT_READY
,
.
ahandle
=
pMsg
->
ahandle
};
SRpcMsg
rspMsg
=
{
.
handle
=
pMsg
->
handle
,
.
code
=
TSDB_CODE_APP_NOT_READY
,
.
ahandle
=
pMsg
->
ahandle
,
.
refId
=
pMsg
->
refId
};
rpcSendResponse
(
&
rspMsg
);
}
rpcFreeCont
(
pMsg
->
pCont
);
...
...
@@ -143,7 +148,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
if
(
isReq
&&
pMsg
->
pCont
==
NULL
)
{
dError
(
"req:%s not processed since its empty, handle:%p app:%p"
,
TMSG_INFO
(
msgType
),
pMsg
->
handle
,
pMsg
->
ahandle
);
SRpcMsg
rspMsg
=
{.
handle
=
pMsg
->
handle
,
.
code
=
TSDB_CODE_INVALID_MSG_LEN
,
.
ahandle
=
pMsg
->
ahandle
};
SRpcMsg
rspMsg
=
{
.
handle
=
pMsg
->
handle
,
.
code
=
TSDB_CODE_INVALID_MSG_LEN
,
.
ahandle
=
pMsg
->
ahandle
,
.
refId
=
pMsg
->
refId
};
rpcSendResponse
(
&
rspMsg
);
return
;
}
...
...
@@ -151,7 +157,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
if
(
pWrapper
==
NULL
)
{
dError
(
"msg:%s not processed since no handle, handle:%p app:%p"
,
TMSG_INFO
(
msgType
),
pMsg
->
handle
,
pMsg
->
ahandle
);
if
(
isReq
)
{
SRpcMsg
rspMsg
=
{.
handle
=
pMsg
->
handle
,
.
code
=
TSDB_CODE_MSG_NOT_PROCESSED
,
.
ahandle
=
pMsg
->
ahandle
};
SRpcMsg
rspMsg
=
{
.
handle
=
pMsg
->
handle
,
.
code
=
TSDB_CODE_MSG_NOT_PROCESSED
,
.
ahandle
=
pMsg
->
ahandle
,
.
refId
=
pMsg
->
refId
};
rpcSendResponse
(
&
rspMsg
);
}
rpcFreeCont
(
pMsg
->
pCont
);
...
...
@@ -170,6 +177,9 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
}
dTrace
(
"msg:%s will be processed by %s, app:%p"
,
TMSG_INFO
(
msgType
),
pWrapper
->
name
,
pMsg
->
ahandle
);
if
(
isReq
)
{
assert
(
pMsg
->
refId
!=
0
);
}
dmProcessRpcMsg
(
pWrapper
,
pMsg
,
pEpSet
);
}
...
...
@@ -317,7 +327,7 @@ static void dmConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t
if
(
code
!=
0
)
{
dError
(
"msg:%p, failed to process since code:0x%04x:%s"
,
pMsg
,
code
&
0XFFFF
,
tstrerror
(
code
));
if
(
pRpc
->
msgType
&
1U
)
{
SRpcMsg
rsp
=
{.
handle
=
pRpc
->
handle
,
.
ahandle
=
pRpc
->
ahandle
,
.
code
=
terrno
};
SRpcMsg
rsp
=
{.
handle
=
pRpc
->
handle
,
.
ahandle
=
pRpc
->
ahandle
,
.
code
=
terrno
,
.
refId
=
pRpc
->
refId
};
dmSendRsp
(
pWrapper
,
&
rsp
);
}
...
...
source/dnode/mgmt/implement/src/dmWorker.c
浏览文件 @
76acde1e
...
...
@@ -105,7 +105,7 @@ void dmStopMonitorThread(SDnode *pDnode) {
}
static
void
dmProcessMgmtQueue
(
SQueueInfo
*
pInfo
,
SNodeMsg
*
pMsg
)
{
SDnode
*
pDnode
=
pInfo
->
ahandle
;
SDnode
*
pDnode
=
pInfo
->
ahandle
;
SRpcMsg
*
pRpc
=
&
pMsg
->
rpcMsg
;
int32_t
code
=
-
1
;
dTrace
(
"msg:%p, will be processed in dnode-mgmt queue"
,
pMsg
);
...
...
@@ -150,7 +150,7 @@ static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
if
(
pRpc
->
msgType
&
1u
)
{
if
(
code
!=
0
)
code
=
terrno
;
SRpcMsg
rsp
=
{.
handle
=
pRpc
->
handle
,
.
ahandle
=
pRpc
->
ahandle
,
.
code
=
code
};
SRpcMsg
rsp
=
{.
handle
=
pRpc
->
handle
,
.
ahandle
=
pRpc
->
ahandle
,
.
code
=
code
,
.
refId
=
pRpc
->
refId
};
rpcSendResponse
(
&
rsp
);
}
...
...
source/dnode/mgmt/interface/src/dmInt.c
浏览文件 @
76acde1e
...
...
@@ -177,7 +177,7 @@ void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) {
SServerStatusRsp
statusRsp
=
{
0
};
dmGetServerStatus
(
pDnode
,
&
statusRsp
);
SRpcMsg
rspMsg
=
{.
handle
=
pReq
->
handle
,
.
ahandle
=
pReq
->
ahandle
};
SRpcMsg
rspMsg
=
{.
handle
=
pReq
->
handle
,
.
ahandle
=
pReq
->
ahandle
,
.
refId
=
pReq
->
refId
};
int32_t
rspLen
=
tSerializeSServerStatusRsp
(
NULL
,
0
,
&
statusRsp
);
if
(
rspLen
<
0
)
{
rspMsg
.
code
=
TSDB_CODE_OUT_OF_MEMORY
;
...
...
source/dnode/mgmt/mgmt_bnode/src/bmWorker.c
浏览文件 @
76acde1e
...
...
@@ -17,7 +17,8 @@
#include "bmInt.h"
static
void
bmSendErrorRsp
(
SNodeMsg
*
pMsg
,
int32_t
code
)
{
SRpcMsg
rpcRsp
=
{.
handle
=
pMsg
->
rpcMsg
.
handle
,
.
ahandle
=
pMsg
->
rpcMsg
.
ahandle
,
.
code
=
code
};
SRpcMsg
rpcRsp
=
{
.
handle
=
pMsg
->
rpcMsg
.
handle
,
.
ahandle
=
pMsg
->
rpcMsg
.
ahandle
,
.
code
=
code
,
.
refId
=
pMsg
->
rpcMsg
.
refId
};
tmsgSendRsp
(
&
rpcRsp
);
dTrace
(
"msg:%p, is freed"
,
pMsg
);
...
...
@@ -38,6 +39,7 @@ static void bmSendErrorRsps(STaosQall *qall, int32_t numOfMsgs, int32_t code) {
static
inline
void
bmSendRsp
(
SNodeMsg
*
pMsg
,
int32_t
code
)
{
SRpcMsg
rsp
=
{.
handle
=
pMsg
->
rpcMsg
.
handle
,
.
ahandle
=
pMsg
->
rpcMsg
.
ahandle
,
.
refId
=
pMsg
->
rpcMsg
.
refId
,
.
code
=
code
,
.
pCont
=
pMsg
->
pRsp
,
.
contLen
=
pMsg
->
rspLen
};
...
...
@@ -101,7 +103,7 @@ static void bmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
}
int32_t
bmProcessWriteMsg
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SBnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SBnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SMultiWorker
*
pWorker
=
&
pMgmt
->
writeWorker
;
dTrace
(
"msg:%p, put into worker:%s"
,
pMsg
,
pWorker
->
name
);
...
...
@@ -110,7 +112,7 @@ int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
}
int32_t
bmProcessMonitorMsg
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SBnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SBnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SSingleWorker
*
pWorker
=
&
pMgmt
->
monitorWorker
;
dTrace
(
"msg:%p, put into worker:%s"
,
pMsg
,
pWorker
->
name
);
...
...
source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
浏览文件 @
76acde1e
...
...
@@ -19,6 +19,7 @@
static
inline
void
mmSendRsp
(
SNodeMsg
*
pMsg
,
int32_t
code
)
{
SRpcMsg
rsp
=
{.
handle
=
pMsg
->
rpcMsg
.
handle
,
.
ahandle
=
pMsg
->
rpcMsg
.
ahandle
,
.
refId
=
pMsg
->
rpcMsg
.
refId
,
.
code
=
code
,
.
pCont
=
pMsg
->
pRsp
,
.
contLen
=
pMsg
->
rspLen
};
...
...
source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
浏览文件 @
76acde1e
...
...
@@ -19,6 +19,7 @@
static
inline
void
qmSendRsp
(
SNodeMsg
*
pMsg
,
int32_t
code
)
{
SRpcMsg
rsp
=
{.
handle
=
pMsg
->
rpcMsg
.
handle
,
.
ahandle
=
pMsg
->
rpcMsg
.
ahandle
,
.
refId
=
pMsg
->
rpcMsg
.
refId
,
.
code
=
code
,
.
pCont
=
pMsg
->
pRsp
,
.
contLen
=
pMsg
->
rspLen
};
...
...
source/dnode/mgmt/mgmt_snode/src/smWorker.c
浏览文件 @
76acde1e
...
...
@@ -19,6 +19,7 @@
static
inline
void
smSendRsp
(
SNodeMsg
*
pMsg
,
int32_t
code
)
{
SRpcMsg
rsp
=
{.
handle
=
pMsg
->
rpcMsg
.
handle
,
.
ahandle
=
pMsg
->
rpcMsg
.
ahandle
,
.
refId
=
pMsg
->
rpcMsg
.
refId
,
.
code
=
code
,
.
pCont
=
pMsg
->
pRsp
,
.
contLen
=
pMsg
->
rspLen
};
...
...
@@ -149,7 +150,7 @@ static FORCE_INLINE int32_t smGetSWTypeFromMsg(SRpcMsg *pMsg) {
}
int32_t
smProcessMgmtMsg
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SSnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SSnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SMultiWorker
*
pWorker
=
taosArrayGetP
(
pMgmt
->
uniqueWorkers
,
0
);
if
(
pWorker
==
NULL
)
{
terrno
=
TSDB_CODE_INVALID_MSG
;
...
...
@@ -162,7 +163,7 @@ int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
}
int32_t
smProcessMonitorMsg
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SSnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SSnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SSingleWorker
*
pWorker
=
&
pMgmt
->
monitorWorker
;
dTrace
(
"msg:%p, put into worker:%s"
,
pMsg
,
pWorker
->
name
);
...
...
@@ -171,7 +172,7 @@ int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
}
int32_t
smProcessUniqueMsg
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SSnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SSnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
int32_t
index
=
smGetSWIdFromMsg
(
&
pMsg
->
rpcMsg
);
SMultiWorker
*
pWorker
=
taosArrayGetP
(
pMgmt
->
uniqueWorkers
,
index
);
if
(
pWorker
==
NULL
)
{
...
...
@@ -185,7 +186,7 @@ int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
}
int32_t
smProcessSharedMsg
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SSnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SSnodeMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SSingleWorker
*
pWorker
=
&
pMgmt
->
sharedWorker
;
dTrace
(
"msg:%p, put into worker:%s"
,
pMsg
,
pWorker
->
name
);
...
...
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
浏览文件 @
76acde1e
...
...
@@ -23,6 +23,7 @@
static
inline
void
vmSendRsp
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
,
int32_t
code
)
{
SRpcMsg
rsp
=
{.
handle
=
pMsg
->
rpcMsg
.
handle
,
.
ahandle
=
pMsg
->
rpcMsg
.
ahandle
,
.
refId
=
pMsg
->
rpcMsg
.
refId
,
.
code
=
code
,
.
pCont
=
pMsg
->
pRsp
,
.
contLen
=
pMsg
->
rspLen
};
...
...
@@ -126,6 +127,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
rsp.code = 0;
rsp.handle = pRpc->handle;
rsp.ahandle = pRpc->ahandle;
rsp.refId = pRpc->refId;
int32_t code = vnodeProcessWriteReq(pVnode->pImpl, pRpc, version++, &rsp);
tmsgSendRsp(&rsp);
...
...
@@ -134,13 +136,14 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
// sync integration response
for
(
int
i
=
0
;
i
<
taosArrayGetSize
(
pArray
);
i
++
)
{
SNodeMsg
*
pMsg
;
SRpcMsg
*
pRpc
;
SRpcMsg
*
pRpc
;
pMsg
=
*
(
SNodeMsg
**
)
taosArrayGet
(
pArray
,
i
);
pRpc
=
&
pMsg
->
rpcMsg
;
rsp
.
ahandle
=
pRpc
->
ahandle
;
rsp
.
handle
=
pRpc
->
handle
;
rsp
.
refId
=
pRpc
->
refId
;
rsp
.
pCont
=
NULL
;
rsp
.
contLen
=
0
;
...
...
@@ -172,11 +175,9 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
static
void
vmProcessApplyQueue
(
SQueueInfo
*
pInfo
,
STaosQall
*
qall
,
int32_t
numOfMsgs
)
{
SVnodeObj
*
pVnode
=
pInfo
->
ahandle
;
SNodeMsg
*
pMsg
=
NULL
;
SNodeMsg
*
pMsg
=
NULL
;
SRpcMsg
rsp
;
// static int64_t version = 0;
for
(
int32_t
i
=
0
;
i
<
numOfMsgs
;
++
i
)
{
#if 1
// sync integration
...
...
@@ -208,6 +209,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
if
(
pMsg
->
rpcMsg
.
handle
!=
NULL
&&
pMsg
->
rpcMsg
.
ahandle
!=
NULL
)
{
rsp
.
ahandle
=
pMsg
->
rpcMsg
.
ahandle
;
rsp
.
handle
=
pMsg
->
rpcMsg
.
handle
;
rsp
.
refId
=
pMsg
->
rpcMsg
.
refId
;
tmsgSendRsp
(
&
rsp
);
}
#endif
...
...
@@ -216,7 +218,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
static
void
vmProcessSyncQueue
(
SQueueInfo
*
pInfo
,
STaosQall
*
qall
,
int32_t
numOfMsgs
)
{
SVnodeObj
*
pVnode
=
pInfo
->
ahandle
;
SNodeMsg
*
pMsg
=
NULL
;
SNodeMsg
*
pMsg
=
NULL
;
for
(
int32_t
i
=
0
;
i
<
numOfMsgs
;
++
i
)
{
taosGetQitem
(
qall
,
(
void
**
)
&
pMsg
);
...
...
@@ -229,7 +231,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
static
void
vmProcessMergeQueue
(
SQueueInfo
*
pInfo
,
STaosQall
*
qall
,
int32_t
numOfMsgs
)
{
SVnodeObj
*
pVnode
=
pInfo
->
ahandle
;
SNodeMsg
*
pMsg
=
NULL
;
SNodeMsg
*
pMsg
=
NULL
;
for
(
int32_t
i
=
0
;
i
<
numOfMsgs
;
++
i
)
{
taosGetQitem
(
qall
,
(
void
**
)
&
pMsg
);
...
...
@@ -246,7 +248,7 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
}
static
int32_t
vmPutNodeMsgToQueue
(
SVnodesMgmt
*
pMgmt
,
SNodeMsg
*
pMsg
,
EQueueType
qtype
)
{
SRpcMsg
*
pRpc
=
&
pMsg
->
rpcMsg
;
SRpcMsg
*
pRpc
=
&
pMsg
->
rpcMsg
;
SMsgHead
*
pHead
=
pRpc
->
pCont
;
pHead
->
contLen
=
ntohl
(
pHead
->
contLen
);
pHead
->
vgId
=
ntohl
(
pHead
->
vgId
);
...
...
@@ -315,7 +317,7 @@ int32_t vmProcessMergeMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
}
int32_t
vmProcessMgmtMsg
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SVnodesMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SVnodesMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SSingleWorker
*
pWorker
=
&
pMgmt
->
mgmtWorker
;
dTrace
(
"msg:%p, will be written to vnode-mgmt queue, worker:%s"
,
pMsg
,
pWorker
->
name
);
taosWriteQitem
(
pWorker
->
queue
,
pMsg
);
...
...
@@ -323,7 +325,7 @@ int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
}
int32_t
vmProcessMonitorMsg
(
SMgmtWrapper
*
pWrapper
,
SNodeMsg
*
pMsg
)
{
SVnodesMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SVnodesMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SSingleWorker
*
pWorker
=
&
pMgmt
->
monitorWorker
;
dTrace
(
"msg:%p, put into worker:%s"
,
pMsg
,
pWorker
->
name
);
...
...
@@ -333,7 +335,7 @@ int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
static
int32_t
vmPutRpcMsgToQueue
(
SMgmtWrapper
*
pWrapper
,
SRpcMsg
*
pRpc
,
EQueueType
qtype
)
{
SVnodesMgmt
*
pMgmt
=
pWrapper
->
pMgmt
;
SMsgHead
*
pHead
=
pRpc
->
pCont
;
SMsgHead
*
pHead
=
pRpc
->
pCont
;
SVnodeObj
*
pVnode
=
vmAcquireVnode
(
pMgmt
,
pHead
->
vgId
);
if
(
pVnode
==
NULL
)
return
-
1
;
...
...
@@ -346,6 +348,7 @@ static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueT
}
else
{
dTrace
(
"msg:%p, is created, type:%s"
,
pMsg
,
TMSG_INFO
(
pRpc
->
msgType
));
pMsg
->
rpcMsg
=
*
pRpc
;
// if (pMsg->rpcMsg.handle != NULL) assert(pMsg->rpcMsg.refId != 0);
switch
(
qtype
)
{
case
QUERY_QUEUE
:
dTrace
(
"msg:%p, will be put into vnode-query queue"
,
pMsg
);
...
...
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @
76acde1e
...
...
@@ -135,6 +135,7 @@ typedef struct {
int32_t
failedTimes
;
void
*
rpcHandle
;
void
*
rpcAHandle
;
int64_t
rpcRefId
;
void
*
rpcRsp
;
int32_t
rpcRspLen
;
SArray
*
redoLogs
;
...
...
source/dnode/mnode/impl/src/mndTrans.c
浏览文件 @
76acde1e
...
...
@@ -193,9 +193,9 @@ TRANS_ENCODE_OVER:
static
SSdbRow
*
mndTransActionDecode
(
SSdbRaw
*
pRaw
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
SSdbRow
*
pRow
=
NULL
;
STrans
*
pTrans
=
NULL
;
char
*
pData
=
NULL
;
SSdbRow
*
pRow
=
NULL
;
STrans
*
pTrans
=
NULL
;
char
*
pData
=
NULL
;
int32_t
dataLen
=
0
;
int8_t
sver
=
0
;
int32_t
redoLogNum
=
0
;
...
...
@@ -456,7 +456,7 @@ static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
}
static
STrans
*
mndAcquireTrans
(
SMnode
*
pMnode
,
int32_t
transId
)
{
SSdb
*
pSdb
=
pMnode
->
pSdb
;
SSdb
*
pSdb
=
pMnode
->
pSdb
;
STrans
*
pTrans
=
sdbAcquire
(
pSdb
,
SDB_TRANS
,
&
transId
);
if
(
pTrans
==
NULL
)
{
terrno
=
TSDB_CODE_MND_TRANS_NOT_EXIST
;
...
...
@@ -484,6 +484,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnType type, const S
pTrans
->
createdTime
=
taosGetTimestampMs
();
pTrans
->
rpcHandle
=
pReq
->
handle
;
pTrans
->
rpcAHandle
=
pReq
->
ahandle
;
pTrans
->
rpcRefId
=
pReq
->
refId
;
pTrans
->
redoLogs
=
taosArrayInit
(
MND_TRANS_ARRAY_SIZE
,
sizeof
(
void
*
));
pTrans
->
undoLogs
=
taosArrayInit
(
MND_TRANS_ARRAY_SIZE
,
sizeof
(
void
*
));
pTrans
->
commitLogs
=
taosArrayInit
(
MND_TRANS_ARRAY_SIZE
,
sizeof
(
void
*
));
...
...
@@ -625,7 +626,7 @@ static int32_t mndCheckTransCanBeStartedInParallel(SMnode *pMnode, STrans *pNewT
if
(
mndIsBasicTrans
(
pNewTrans
))
return
0
;
STrans
*
pTrans
=
NULL
;
void
*
pIter
=
NULL
;
void
*
pIter
=
NULL
;
int32_t
code
=
0
;
while
(
1
)
{
...
...
@@ -703,6 +704,7 @@ int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
pNew
->
rpcHandle
=
pTrans
->
rpcHandle
;
pNew
->
rpcAHandle
=
pTrans
->
rpcAHandle
;
pNew
->
rpcRefId
=
pTrans
->
rpcRefId
;
pNew
->
rpcRsp
=
pTrans
->
rpcRsp
;
pNew
->
rpcRspLen
=
pTrans
->
rpcRspLen
;
pTrans
->
rpcRsp
=
NULL
;
...
...
@@ -767,6 +769,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
SRpcMsg
rspMsg
=
{.
handle
=
pTrans
->
rpcHandle
,
.
code
=
pTrans
->
code
,
.
ahandle
=
pTrans
->
rpcAHandle
,
.
refId
=
pTrans
->
rpcRefId
,
.
pCont
=
rpcCont
,
.
contLen
=
pTrans
->
rpcRspLen
};
tmsgSendRsp
(
&
rspMsg
);
...
...
@@ -827,7 +830,7 @@ HANDLE_ACTION_RSP_OVER:
}
static
int32_t
mndTransExecuteLogs
(
SMnode
*
pMnode
,
SArray
*
pArray
)
{
SSdb
*
pSdb
=
pMnode
->
pSdb
;
SSdb
*
pSdb
=
pMnode
->
pSdb
;
int32_t
arraySize
=
taosArrayGetSize
(
pArray
);
if
(
arraySize
==
0
)
return
0
;
...
...
@@ -1202,11 +1205,11 @@ static int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
}
static
int32_t
mndProcessKillTransReq
(
SNodeMsg
*
pReq
)
{
SMnode
*
pMnode
=
pReq
->
pNode
;
SMnode
*
pMnode
=
pReq
->
pNode
;
SKillTransReq
killReq
=
{
0
};
int32_t
code
=
-
1
;
SUserObj
*
pUser
=
NULL
;
STrans
*
pTrans
=
NULL
;
SUserObj
*
pUser
=
NULL
;
STrans
*
pTrans
=
NULL
;
if
(
tDeserializeSKillTransReq
(
pReq
->
rpcMsg
.
pCont
,
pReq
->
rpcMsg
.
contLen
,
&
killReq
)
!=
0
)
{
terrno
=
TSDB_CODE_INVALID_MSG
;
...
...
@@ -1246,7 +1249,7 @@ KILL_OVER:
void
mndTransPullup
(
SMnode
*
pMnode
)
{
STrans
*
pTrans
=
NULL
;
void
*
pIter
=
NULL
;
void
*
pIter
=
NULL
;
while
(
1
)
{
pIter
=
sdbFetch
(
pMnode
->
pSdb
,
SDB_TRANS
,
pIter
,
(
void
**
)
&
pTrans
);
...
...
@@ -1261,11 +1264,11 @@ void mndTransPullup(SMnode *pMnode) {
static
int32_t
mndRetrieveTrans
(
SNodeMsg
*
pReq
,
SShowObj
*
pShow
,
SSDataBlock
*
pBlock
,
int32_t
rows
)
{
SMnode
*
pMnode
=
pReq
->
pNode
;
SSdb
*
pSdb
=
pMnode
->
pSdb
;
SSdb
*
pSdb
=
pMnode
->
pSdb
;
int32_t
numOfRows
=
0
;
STrans
*
pTrans
=
NULL
;
int32_t
cols
=
0
;
char
*
pWrite
;
char
*
pWrite
;
while
(
numOfRows
<
rows
)
{
pShow
->
pIter
=
sdbFetch
(
pSdb
,
SDB_TRANS
,
pShow
->
pIter
,
(
void
**
)
&
pTrans
);
...
...
source/dnode/vnode/src/vnd/vnodeQuery.c
浏览文件 @
76acde1e
...
...
@@ -22,21 +22,21 @@ int vnodeQueryOpen(SVnode *pVnode) {
void
vnodeQueryClose
(
SVnode
*
pVnode
)
{
qWorkerDestroy
((
void
**
)
&
pVnode
->
pQuery
);
}
int
vnodeGetTableMeta
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
)
{
STbCfg
*
pTbCfg
=
NULL
;
STbCfg
*
pStbCfg
=
NULL
;
STbCfg
*
pTbCfg
=
NULL
;
STbCfg
*
pStbCfg
=
NULL
;
tb_uid_t
uid
;
int32_t
nCols
;
int32_t
nTagCols
;
SSchemaWrapper
*
pSW
=
NULL
;
STableMetaRsp
*
pTbMetaMsg
=
NULL
;
STableMetaRsp
*
pTbMetaMsg
=
NULL
;
STableMetaRsp
metaRsp
=
{
0
};
SSchema
*
pTagSchema
;
SSchema
*
pTagSchema
;
SRpcMsg
rpcMsg
;
int
msgLen
=
0
;
int32_t
code
=
0
;
char
tableFName
[
TSDB_TABLE_FNAME_LEN
];
int32_t
rspLen
=
0
;
void
*
pRsp
=
NULL
;
void
*
pRsp
=
NULL
;
STableInfoReq
infoReq
=
{
0
};
if
(
tDeserializeSTableInfoReq
(
pMsg
->
pCont
,
pMsg
->
contLen
,
&
infoReq
)
!=
0
)
{
...
...
@@ -142,6 +142,7 @@ _exit:
rpcMsg
.
handle
=
pMsg
->
handle
;
rpcMsg
.
ahandle
=
pMsg
->
ahandle
;
rpcMsg
.
refId
=
pMsg
->
refId
;
rpcMsg
.
pCont
=
pRsp
;
rpcMsg
.
contLen
=
rspLen
;
rpcMsg
.
code
=
code
;
...
...
source/dnode/vnode/src/vnd/vnodeSync.c
浏览文件 @
76acde1e
...
...
@@ -113,7 +113,7 @@ void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb
pFsm
,
cbMeta
.
index
,
cbMeta
.
isWeak
,
cbMeta
.
code
,
cbMeta
.
state
,
syncUtilState2String
(
cbMeta
.
state
),
beginIndex
);
syncRpcMsgLog2
(
logBuf
,
(
SRpcMsg
*
)
pMsg
);
SVnode
*
pVnode
=
(
SVnode
*
)(
pFsm
->
data
);
SVnode
*
pVnode
=
(
SVnode
*
)(
pFsm
->
data
);
SyncApplyMsg
*
pSyncApplyMsg
=
syncApplyMsgBuild2
(
pMsg
,
pVnode
->
config
.
vgId
,
&
cbMeta
);
SRpcMsg
applyMsg
;
syncApplyMsg2RpcMsg
(
pSyncApplyMsg
,
&
applyMsg
);
...
...
@@ -133,6 +133,7 @@ void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb
if
(
ret
==
1
&&
cbMeta
.
state
==
TAOS_SYNC_STATE_LEADER
)
{
applyMsg
.
handle
=
saveRpcMsg
.
handle
;
applyMsg
.
ahandle
=
saveRpcMsg
.
ahandle
;
applyMsg
.
refId
=
saveRpcMsg
.
refId
;
}
else
{
applyMsg
.
handle
=
NULL
;
applyMsg
.
ahandle
=
NULL
;
...
...
source/libs/executor/src/executorimpl.c
浏览文件 @
76acde1e
此差异已折叠。
点击以展开。
source/libs/qworker/inc/qworkerInt.h
浏览文件 @
76acde1e
...
...
@@ -60,7 +60,6 @@ enum {
QW_WRITE
,
};
enum
{
QW_NOT_EXIST_RET_ERR
=
1
,
QW_NOT_EXIST_ADD
,
...
...
@@ -73,14 +72,15 @@ typedef struct SQWDebug {
}
SQWDebug
;
typedef
struct
SQWConnInfo
{
void
*
handle
;
void
*
ahandle
;
void
*
handle
;
void
*
ahandle
;
int64_t
refId
;
}
SQWConnInfo
;
typedef
struct
SQWMsg
{
void
*
node
;
void
*
node
;
int32_t
code
;
char
*
msg
;
char
*
msg
;
int32_t
msgLen
;
SQWConnInfo
connInfo
;
}
SQWMsg
;
...
...
@@ -100,7 +100,6 @@ typedef struct SQWPhaseOutput {
#endif
}
SQWPhaseOutput
;
typedef
struct
SQWTaskStatus
{
int64_t
refId
;
// job's refId
int32_t
code
;
...
...
@@ -124,8 +123,8 @@ typedef struct SQWTaskCtx {
int8_t
events
[
QW_EVENT_MAX
];
qTaskInfo_t
taskHandle
;
DataSinkHandle
sinkHandle
;
void
*
taskHandle
;
void
*
sinkHandle
;
}
SQWTaskCtx
;
typedef
struct
SQWSchStatus
{
...
...
@@ -134,7 +133,7 @@ typedef struct SQWSchStatus {
SQWConnInfo
hbConnInfo
;
SQueryNodeEpId
hbEpId
;
SRWLatch
tasksLock
;
SHashObj
*
tasksHash
;
// key:queryId+taskId, value: SQWTaskStatus
SHashObj
*
tasksHash
;
// key:queryId+taskId, value: SQWTaskStatus
}
SQWSchStatus
;
// Qnode/Vnode level task management
...
...
@@ -142,7 +141,7 @@ typedef struct SQWorkerMgmt {
SQWorkerCfg
cfg
;
int8_t
nodeType
;
int32_t
nodeId
;
void
*
timer
;
void
*
timer
;
tmr_h
hbTimer
;
SRWLatch
schLock
;
// SRWLatch ctxLock;
...
...
@@ -171,41 +170,86 @@ typedef struct SQWorkerMgmt {
#define QW_TASK_NOT_EXIST(code) (TSDB_CODE_QRY_SCH_NOT_EXIST == (code) || TSDB_CODE_QRY_TASK_NOT_EXIST == (code))
#define QW_TASK_ALREADY_EXIST(code) (TSDB_CODE_QRY_TASK_ALREADY_EXIST == (code))
#define QW_TASK_READY(status) (status == JOB_TASK_STATUS_SUCCEED || status == JOB_TASK_STATUS_FAILED || status == JOB_TASK_STATUS_CANCELLED || status == JOB_TASK_STATUS_PARTIAL_SUCCEED)
#define QW_SET_QTID(id, qId, tId) do { *(uint64_t *)(id) = (qId); *(uint64_t *)((char *)(id) + sizeof(qId)) = (tId); } while (0)
#define QW_GET_QTID(id, qId, tId) do { (qId) = *(uint64_t *)(id); (tId) = *(uint64_t *)((char *)(id) + sizeof(qId)); } while (0)
#define QW_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
#define QW_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define QW_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
#define QW_TASK_READY(status) \
(status == JOB_TASK_STATUS_SUCCEED || status == JOB_TASK_STATUS_FAILED || status == JOB_TASK_STATUS_CANCELLED || \
status == JOB_TASK_STATUS_PARTIAL_SUCCEED)
#define QW_SET_QTID(id, qId, tId) \
do { \
*(uint64_t *)(id) = (qId); \
*(uint64_t *)((char *)(id) + sizeof(qId)) = (tId); \
} while (0)
#define QW_GET_QTID(id, qId, tId) \
do { \
(qId) = *(uint64_t *)(id); \
(tId) = *(uint64_t *)((char *)(id) + sizeof(qId)); \
} while (0)
#define QW_ERR_RET(c) \
do { \
int32_t _code = c; \
if (_code != TSDB_CODE_SUCCESS) { \
terrno = _code; \
return _code; \
} \
} while (0)
#define QW_RET(c) \
do { \
int32_t _code = c; \
if (_code != TSDB_CODE_SUCCESS) { \
terrno = _code; \
} \
return _code; \
} while (0)
#define QW_ERR_JRET(c) \
do { \
code = c; \
if (code != TSDB_CODE_SUCCESS) { \
terrno = code; \
goto _return; \
} \
} while (0)
#define QW_ELOG(param, ...) qError("QW:%p " param, mgmt, __VA_ARGS__)
#define QW_DLOG(param, ...) qDebug("QW:%p " param, mgmt, __VA_ARGS__)
#define QW_DUMP(param, ...) do { if (gQWDebug.dumpEnable) { qDebug("QW:%p " param, mgmt, __VA_ARGS__); } } while (0)
#define QW_SCH_ELOG(param, ...) qError("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__)
#define QW_SCH_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__)
#define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOGL(param, ...) qDebugL("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
#define QW_SCH_TASK_ELOG(param, ...) qError("QW:%p SID:0x%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_WLOG(param, ...) qWarn("QW:%p SID:0x%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_DLOG(param, ...) qDebug("QW:%p SID:0x%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_LOCK_DEBUG(...) do { if (gQWDebug.lockEnable) { qDebug(__VA_ARGS__); } } while (0)
#define QW_DUMP(param, ...) \
do { \
if (gQWDebug.dumpEnable) { \
qDebug("QW:%p " param, mgmt, __VA_ARGS__); \
} \
} while (0)
#define QW_SCH_ELOG(param, ...) qError("QW:%p SID:%" PRIx64 " " param, mgmt, sId, __VA_ARGS__)
#define QW_SCH_DLOG(param, ...) qDebug("QW:%p SID:%" PRIx64 " " param, mgmt, sId, __VA_ARGS__)
#define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOGL(param, ...) \
qDebugL("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId)
#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId)
#define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId)
#define QW_SCH_TASK_ELOG(param, ...) \
qError("QW:%p SID:0x%" PRIx64 ",QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_WLOG(param, ...) \
qWarn("QW:%p SID:0x%" PRIx64 ",QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_DLOG(param, ...) \
qDebug("QW:%p SID:0x%" PRIx64 ",QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_LOCK_DEBUG(...) \
do { \
if (gQWDebug.lockEnable) { \
qDebug(__VA_ARGS__); \
} \
} while (0)
#define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000
#define QW_LOCK(type, _lock) do { \
#define QW_LOCK(type, _lock) \
do { \
if (QW_READ == (type)) { \
assert(atomic_load_32((_lock)) >= 0); \
QW_LOCK_DEBUG("QW RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
...
...
@@ -219,9 +263,10 @@ typedef struct SQWorkerMgmt {
QW_LOCK_DEBUG("QW WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY); \
} \
} while (0)
} while (0)
#define QW_UNLOCK(type, _lock) do { \
#define QW_UNLOCK(type, _lock) \
do { \
if (QW_READ == (type)) { \
assert(atomic_load_32((_lock)) > 0); \
QW_LOCK_DEBUG("QW RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
...
...
@@ -235,7 +280,7 @@ typedef struct SQWorkerMgmt {
QW_LOCK_DEBUG("QW WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
assert(atomic_load_32((_lock)) >= 0); \
} \
} while (0)
} while (0)
#ifdef __cplusplus
}
...
...
source/libs/qworker/src/qworker.c
浏览文件 @
76acde1e
此差异已折叠。
点击以展开。
source/libs/qworker/src/qworkerMsg.c
浏览文件 @
76acde1e
#include "qworker.h"
#include "
tcommon
.h"
#include "qworker
Msg
.h"
#include "
dataSinkMgt
.h"
#include "executor.h"
#include "planner.h"
#include "query.h"
#include "qworker.h"
#include "qworkerInt.h"
#include "
qworkerMsg
.h"
#include "
tcommon
.h"
#include "tmsg.h"
#include "tname.h"
#include "dataSinkMgt.h"
int32_t
qwMallocFetchRsp
(
int32_t
length
,
SRetrieveTableRsp
**
rsp
)
{
int32_t
msgSize
=
sizeof
(
SRetrieveTableRsp
)
+
length
;
...
...
@@ -26,8 +25,6 @@ int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp) {
return
TSDB_CODE_SUCCESS
;
}
void
qwBuildFetchRsp
(
void
*
msg
,
SOutputData
*
input
,
int32_t
len
,
bool
qComplete
)
{
SRetrieveTableRsp
*
rsp
=
(
SRetrieveTableRsp
*
)
msg
;
...
...
@@ -39,7 +36,6 @@ void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComplete)
rsp
->
numOfRows
=
htonl
(
input
->
numOfRows
);
}
void
qwFreeFetchRsp
(
void
*
msg
)
{
if
(
msg
)
{
rpcFreeCont
(
msg
);
...
...
@@ -50,13 +46,14 @@ int32_t qwBuildAndSendQueryRsp(SQWConnInfo *pConn, int32_t code) {
SQueryTableRsp
rsp
=
{.
code
=
code
};
int32_t
contLen
=
tSerializeSQueryTableRsp
(
NULL
,
0
,
&
rsp
);
void
*
msg
=
rpcMallocCont
(
contLen
);
void
*
msg
=
rpcMallocCont
(
contLen
);
tSerializeSQueryTableRsp
(
msg
,
contLen
,
&
rsp
);
SRpcMsg
rpcRsp
=
{
.
msgType
=
TDMT_VND_QUERY_RSP
,
.
handle
=
pConn
->
handle
,
.
ahandle
=
pConn
->
ahandle
,
.
refId
=
pConn
->
refId
,
.
pCont
=
msg
,
.
contLen
=
contLen
,
.
code
=
code
,
...
...
@@ -74,6 +71,7 @@ int32_t qwBuildAndSendReadyRsp(SQWConnInfo *pConn, int32_t code) {
SRpcMsg
rpcRsp
=
{
.
msgType
=
TDMT_VND_RES_READY_RSP
,
.
handle
=
pConn
->
handle
,
.
refId
=
pConn
->
refId
,
.
ahandle
=
NULL
,
.
pCont
=
pRsp
,
.
contLen
=
sizeof
(
*
pRsp
),
...
...
@@ -89,13 +87,14 @@ int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo,
SExplainRsp
rsp
=
{.
numOfPlans
=
num
,
.
subplanInfo
=
execInfo
};
int32_t
contLen
=
tSerializeSExplainRsp
(
NULL
,
0
,
&
rsp
);
void
*
pRsp
=
rpcMallocCont
(
contLen
);
void
*
pRsp
=
rpcMallocCont
(
contLen
);
tSerializeSExplainRsp
(
pRsp
,
contLen
,
&
rsp
);
SRpcMsg
rpcRsp
=
{
.
msgType
=
TDMT_VND_EXPLAIN_RSP
,
.
handle
=
pConn
->
handle
,
.
ahandle
=
pConn
->
ahandle
,
.
refId
=
pConn
->
refId
,
.
pCont
=
pRsp
,
.
contLen
=
contLen
,
.
code
=
0
,
...
...
@@ -108,13 +107,14 @@ int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo,
int32_t
qwBuildAndSendHbRsp
(
SQWConnInfo
*
pConn
,
SSchedulerHbRsp
*
pStatus
,
int32_t
code
)
{
int32_t
contLen
=
tSerializeSSchedulerHbRsp
(
NULL
,
0
,
pStatus
);
void
*
pRsp
=
rpcMallocCont
(
contLen
);
void
*
pRsp
=
rpcMallocCont
(
contLen
);
tSerializeSSchedulerHbRsp
(
pRsp
,
contLen
,
pStatus
);
SRpcMsg
rpcRsp
=
{
.
msgType
=
TDMT_VND_QUERY_HEARTBEAT_RSP
,
.
handle
=
pConn
->
handle
,
.
ahandle
=
pConn
->
ahandle
,
.
refId
=
pConn
->
refId
,
.
pCont
=
pRsp
,
.
contLen
=
contLen
,
.
code
=
code
,
...
...
@@ -136,6 +136,7 @@ int32_t qwBuildAndSendFetchRsp(SQWConnInfo *pConn, SRetrieveTableRsp *pRsp, int3
.
msgType
=
TDMT_VND_FETCH_RSP
,
.
handle
=
pConn
->
handle
,
.
ahandle
=
pConn
->
ahandle
,
.
refId
=
pConn
->
refId
,
.
pCont
=
pRsp
,
.
contLen
=
sizeof
(
*
pRsp
)
+
dataLength
,
.
code
=
code
,
...
...
@@ -154,6 +155,7 @@ int32_t qwBuildAndSendCancelRsp(SQWConnInfo *pConn, int32_t code) {
.
msgType
=
TDMT_VND_CANCEL_TASK_RSP
,
.
handle
=
pConn
->
handle
,
.
ahandle
=
pConn
->
ahandle
,
.
refId
=
pConn
->
refId
,
.
pCont
=
pRsp
,
.
contLen
=
sizeof
(
*
pRsp
),
.
code
=
code
,
...
...
@@ -171,6 +173,7 @@ int32_t qwBuildAndSendDropRsp(SQWConnInfo *pConn, int32_t code) {
.
msgType
=
TDMT_VND_DROP_TASK_RSP
,
.
handle
=
pConn
->
handle
,
.
ahandle
=
pConn
->
ahandle
,
.
refId
=
pConn
->
refId
,
.
pCont
=
pRsp
,
.
contLen
=
sizeof
(
*
pRsp
),
.
code
=
code
,
...
...
@@ -220,12 +223,13 @@ int32_t qwBuildAndSendShowRsp(SRpcMsg *pMsg, int32_t code) {
showRsp
.
tableMeta
.
numOfColumns
=
cols
;
int32_t
bufLen
=
tSerializeSShowRsp
(
NULL
,
0
,
&
showRsp
);
void
*
pBuf
=
rpcMallocCont
(
bufLen
);
void
*
pBuf
=
rpcMallocCont
(
bufLen
);
tSerializeSShowRsp
(
pBuf
,
bufLen
,
&
showRsp
);
SRpcMsg
rpcMsg
=
{
.
handle
=
pMsg
->
handle
,
.
ahandle
=
pMsg
->
ahandle
,
.
refId
=
pMsg
->
refId
,
.
pCont
=
pBuf
,
.
contLen
=
bufLen
,
.
code
=
code
,
...
...
@@ -235,7 +239,7 @@ int32_t qwBuildAndSendShowRsp(SRpcMsg *pMsg, int32_t code) {
return
TSDB_CODE_SUCCESS
;
}
int32_t
qwBuildAndSendShowFetchRsp
(
SRpcMsg
*
pMsg
,
SVShowTablesFetchReq
*
pFetchReq
)
{
int32_t
qwBuildAndSendShowFetchRsp
(
SRpcMsg
*
pMsg
,
SVShowTablesFetchReq
*
pFetchReq
)
{
SVShowTablesFetchRsp
*
pRsp
=
(
SVShowTablesFetchRsp
*
)
rpcMallocCont
(
sizeof
(
SVShowTablesFetchRsp
));
int32_t
handle
=
htonl
(
pFetchReq
->
id
);
...
...
@@ -243,6 +247,7 @@ int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchRe
SRpcMsg
rpcMsg
=
{
.
handle
=
pMsg
->
handle
,
.
ahandle
=
pMsg
->
ahandle
,
.
refId
=
pMsg
->
refId
,
.
pCont
=
pRsp
,
.
contLen
=
sizeof
(
*
pRsp
),
.
code
=
0
,
...
...
@@ -253,7 +258,7 @@ int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchRe
}
int32_t
qwBuildAndSendCQueryMsg
(
QW_FPARAMS_DEF
,
SQWConnInfo
*
pConn
)
{
SQueryContinueReq
*
req
=
(
SQueryContinueReq
*
)
rpcMallocCont
(
sizeof
(
SQueryContinueReq
));
SQueryContinueReq
*
req
=
(
SQueryContinueReq
*
)
rpcMallocCont
(
sizeof
(
SQueryContinueReq
));
if
(
NULL
==
req
)
{
QW_SCH_TASK_ELOG
(
"rpcMallocCont %d failed"
,
(
int32_t
)
sizeof
(
SQueryContinueReq
));
QW_ERR_RET
(
TSDB_CODE_QRY_OUT_OF_MEMORY
);
...
...
@@ -268,6 +273,7 @@ int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
.
handle
=
pConn
->
handle
,
.
ahandle
=
pConn
->
ahandle
,
.
msgType
=
TDMT_VND_QUERY_CONTINUE
,
.
refId
=
pConn
->
refId
,
.
pCont
=
req
,
.
contLen
=
sizeof
(
SQueryContinueReq
),
.
code
=
0
,
...
...
@@ -285,9 +291,8 @@ int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
return
TSDB_CODE_SUCCESS
;
}
int32_t
qwRegisterQueryBrokenLinkArg
(
QW_FPARAMS_DEF
,
SQWConnInfo
*
pConn
)
{
STaskDropReq
*
req
=
(
STaskDropReq
*
)
rpcMallocCont
(
sizeof
(
STaskDropReq
));
STaskDropReq
*
req
=
(
STaskDropReq
*
)
rpcMallocCont
(
sizeof
(
STaskDropReq
));
if
(
NULL
==
req
)
{
QW_SCH_TASK_ELOG
(
"rpcMallocCont %d failed"
,
(
int32_t
)
sizeof
(
STaskDropReq
));
QW_ERR_RET
(
TSDB_CODE_QRY_OUT_OF_MEMORY
);
...
...
@@ -302,6 +307,7 @@ int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
SRpcMsg
pMsg
=
{
.
handle
=
pConn
->
handle
,
.
ahandle
=
pConn
->
ahandle
,
.
refId
=
pConn
->
refId
,
.
msgType
=
TDMT_VND_DROP_TASK
,
.
pCont
=
req
,
.
contLen
=
sizeof
(
STaskDropReq
),
...
...
@@ -337,6 +343,7 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorkerMgmt *mgmt, uint64_t sId, SQWConnInfo
SRpcMsg
pMsg
=
{
.
handle
=
pConn
->
handle
,
.
ahandle
=
pConn
->
ahandle
,
.
refId
=
pConn
->
refId
,
.
msgType
=
TDMT_VND_QUERY_HEARTBEAT
,
.
pCont
=
msg
,
.
contLen
=
sizeof
(
SSchedulerHbReq
),
...
...
@@ -348,8 +355,6 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorkerMgmt *mgmt, uint64_t sId, SQWConnInfo
return
TSDB_CODE_SUCCESS
;
}
int32_t
qWorkerProcessQueryMsg
(
void
*
node
,
void
*
qWorkerMgmt
,
SRpcMsg
*
pMsg
)
{
if
(
NULL
==
node
||
NULL
==
qWorkerMgmt
||
NULL
==
pMsg
)
{
QW_ERR_RET
(
TSDB_CODE_QRY_INVALID_INPUT
);
...
...
@@ -379,8 +384,9 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg
qwMsg
=
{.
node
=
node
,
.
msg
=
msg
->
msg
+
msg
->
sqlLen
,
.
msgLen
=
msg
->
phyLen
};
qwMsg
.
connInfo
.
handle
=
pMsg
->
handle
;
qwMsg
.
connInfo
.
ahandle
=
pMsg
->
ahandle
;
qwMsg
.
connInfo
.
refId
=
pMsg
->
refId
;
char
*
sql
=
strndup
(
msg
->
msg
,
msg
->
sqlLen
);
char
*
sql
=
strndup
(
msg
->
msg
,
msg
->
sqlLen
);
QW_SCH_TASK_DLOG
(
"processQuery start, node:%p, handle:%p, sql:%s"
,
node
,
pMsg
->
handle
,
sql
);
taosMemoryFreeClear
(
sql
);
...
...
@@ -397,8 +403,8 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
bool
queryDone
=
false
;
SQueryContinueReq
*
msg
=
(
SQueryContinueReq
*
)
pMsg
->
pCont
;
bool
needStop
=
false
;
SQWTaskCtx
*
handles
=
NULL
;
SQWorkerMgmt
*
mgmt
=
(
SQWorkerMgmt
*
)
qWorkerMgmt
;
SQWTaskCtx
*
handles
=
NULL
;
SQWorkerMgmt
*
mgmt
=
(
SQWorkerMgmt
*
)
qWorkerMgmt
;
if
(
NULL
==
msg
||
pMsg
->
contLen
<
sizeof
(
*
msg
))
{
QW_ELOG
(
"invalid cquery msg, msg:%p, msgLen:%d"
,
msg
,
pMsg
->
contLen
);
...
...
@@ -413,6 +419,7 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg
qwMsg
=
{.
node
=
node
,
.
msg
=
NULL
,
.
msgLen
=
0
};
qwMsg
.
connInfo
.
handle
=
pMsg
->
handle
;
qwMsg
.
connInfo
.
ahandle
=
pMsg
->
ahandle
;
qwMsg
.
connInfo
.
refId
=
pMsg
->
refId
;
QW_SCH_TASK_DLOG
(
"processCQuery start, node:%p, handle:%p"
,
node
,
pMsg
->
handle
);
...
...
@@ -423,7 +430,7 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
return
TSDB_CODE_SUCCESS
;
}
int32_t
qWorkerProcessReadyMsg
(
void
*
node
,
void
*
qWorkerMgmt
,
SRpcMsg
*
pMsg
){
int32_t
qWorkerProcessReadyMsg
(
void
*
node
,
void
*
qWorkerMgmt
,
SRpcMsg
*
pMsg
)
{
if
(
NULL
==
node
||
NULL
==
qWorkerMgmt
||
NULL
==
pMsg
)
{
return
TSDB_CODE_QRY_INVALID_INPUT
;
}
...
...
@@ -447,6 +454,7 @@ int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){
SQWMsg
qwMsg
=
{.
node
=
node
,
.
msg
=
NULL
,
.
msgLen
=
0
};
qwMsg
.
connInfo
.
handle
=
pMsg
->
handle
;
qwMsg
.
connInfo
.
ahandle
=
pMsg
->
ahandle
;
qwMsg
.
connInfo
.
refId
=
pMsg
->
refId
;
QW_SCH_TASK_DLOG
(
"processReady start, node:%p, handle:%p"
,
node
,
pMsg
->
handle
);
...
...
@@ -475,11 +483,11 @@ int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SSchedulerStatusRsp
*
sStatus
=
NULL
;
//QW_ERR_JRET(qwGetSchTasksStatus(qWorkerMgmt, msg->sId, &sStatus));
//
QW_ERR_JRET(qwGetSchTasksStatus(qWorkerMgmt, msg->sId, &sStatus));
_return:
//QW_ERR_RET(qwBuildAndSendStatusRsp(pMsg, sStatus));
//
QW_ERR_RET(qwBuildAndSendStatusRsp(pMsg, sStatus));
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -509,6 +517,7 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg
qwMsg
=
{.
node
=
node
,
.
msg
=
NULL
,
.
msgLen
=
0
};
qwMsg
.
connInfo
.
handle
=
pMsg
->
handle
;
qwMsg
.
connInfo
.
ahandle
=
pMsg
->
ahandle
;
qwMsg
.
connInfo
.
refId
=
pMsg
->
refId
;
QW_SCH_TASK_DLOG
(
"processFetch start, node:%p, handle:%p"
,
node
,
pMsg
->
handle
);
...
...
@@ -529,7 +538,7 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
return
TSDB_CODE_QRY_INVALID_INPUT
;
}
SQWorkerMgmt
*
mgmt
=
(
SQWorkerMgmt
*
)
qWorkerMgmt
;
SQWorkerMgmt
*
mgmt
=
(
SQWorkerMgmt
*
)
qWorkerMgmt
;
int32_t
code
=
0
;
STaskCancelReq
*
msg
=
pMsg
->
pCont
;
if
(
NULL
==
msg
||
pMsg
->
contLen
<
sizeof
(
*
msg
))
{
...
...
@@ -550,8 +559,9 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg
qwMsg
=
{.
node
=
node
,
.
msg
=
NULL
,
.
msgLen
=
0
};
qwMsg
.
connInfo
.
handle
=
pMsg
->
handle
;
qwMsg
.
connInfo
.
ahandle
=
pMsg
->
ahandle
;
qwMsg
.
connInfo
.
refId
=
pMsg
->
refId
;
//QW_ERR_JRET(qwCancelTask(qWorkerMgmt, msg->sId, msg->queryId, msg->taskId));
//
QW_ERR_JRET(qwCancelTask(qWorkerMgmt, msg->sId, msg->queryId, msg->taskId));
_return:
...
...
@@ -588,6 +598,7 @@ int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg
qwMsg
=
{.
node
=
node
,
.
msg
=
NULL
,
.
msgLen
=
0
,
.
code
=
pMsg
->
code
};
qwMsg
.
connInfo
.
handle
=
pMsg
->
handle
;
qwMsg
.
connInfo
.
ahandle
=
pMsg
->
ahandle
;
qwMsg
.
connInfo
.
refId
=
pMsg
->
refId
;
if
(
TSDB_CODE_RPC_NETWORK_UNAVAIL
==
pMsg
->
code
)
{
QW_SCH_TASK_DLOG
(
"receive drop task due to network broken, error:%s"
,
tstrerror
(
pMsg
->
code
));
...
...
@@ -609,7 +620,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
int32_t
code
=
0
;
SSchedulerHbReq
req
=
{
0
};
SQWorkerMgmt
*
mgmt
=
(
SQWorkerMgmt
*
)
qWorkerMgmt
;
SQWorkerMgmt
*
mgmt
=
(
SQWorkerMgmt
*
)
qWorkerMgmt
;
if
(
NULL
==
pMsg
->
pCont
)
{
QW_ELOG
(
"invalid hb msg, msg:%p, msgLen:%d"
,
pMsg
->
pCont
,
pMsg
->
contLen
);
...
...
@@ -626,6 +637,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg
qwMsg
=
{.
node
=
node
,
.
msg
=
NULL
,
.
msgLen
=
0
,
.
code
=
pMsg
->
code
};
qwMsg
.
connInfo
.
handle
=
pMsg
->
handle
;
qwMsg
.
connInfo
.
ahandle
=
pMsg
->
ahandle
;
qwMsg
.
connInfo
.
refId
=
pMsg
->
refId
;
if
(
TSDB_CODE_RPC_NETWORK_UNAVAIL
==
pMsg
->
code
)
{
QW_SCH_DLOG
(
"receive Hb msg due to network broken, error:%s"
,
tstrerror
(
pMsg
->
code
));
...
...
source/libs/scheduler/src/scheduler.c
浏览文件 @
76acde1e
...
...
@@ -369,7 +369,7 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) {
}
for
(
int32_t
n
=
0
;
n
<
childNum
;
++
n
)
{
SSubplan
*
child
=
(
SSubplan
*
)
nodesListGetNode
(
pPlan
->
pChildren
,
n
);
SSubplan
*
child
=
(
SSubplan
*
)
nodesListGetNode
(
pPlan
->
pChildren
,
n
);
SSchTask
**
childTask
=
taosHashGet
(
planToTask
,
&
child
,
POINTER_BYTES
);
if
(
NULL
==
childTask
||
NULL
==
*
childTask
)
{
SCH_TASK_ELOG
(
"subplan children relationship error, level:%d, taskIdx:%d, childIdx:%d"
,
i
,
m
,
n
);
...
...
@@ -401,7 +401,7 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) {
}
for
(
int32_t
n
=
0
;
n
<
parentNum
;
++
n
)
{
SSubplan
*
parent
=
(
SSubplan
*
)
nodesListGetNode
(
pPlan
->
pParents
,
n
);
SSubplan
*
parent
=
(
SSubplan
*
)
nodesListGetNode
(
pPlan
->
pParents
,
n
);
SSchTask
**
parentTask
=
taosHashGet
(
planToTask
,
&
parent
,
POINTER_BYTES
);
if
(
NULL
==
parentTask
||
NULL
==
*
parentTask
)
{
SCH_TASK_ELOG
(
"subplan parent relationship error, level:%d, taskIdx:%d, childIdx:%d"
,
i
,
m
,
n
);
...
...
@@ -491,7 +491,7 @@ int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) {
SSchLevel
level
=
{
0
};
SNodeListNode
*
plans
=
NULL
;
int32_t
taskNum
=
0
;
SSchLevel
*
pLevel
=
NULL
;
SSchLevel
*
pLevel
=
NULL
;
level
.
status
=
JOB_TASK_STATUS_NOT_START
;
...
...
@@ -1267,7 +1267,7 @@ int32_t schUpdateTaskExecNodeHandle(SSchTask *pTask, void *handle, int32_t rspCo
int32_t
schHandleCallback
(
void
*
param
,
const
SDataBuf
*
pMsg
,
int32_t
msgType
,
int32_t
rspCode
)
{
int32_t
code
=
0
;
SSchTaskCallbackParam
*
pParam
=
(
SSchTaskCallbackParam
*
)
param
;
SSchTask
*
pTask
=
NULL
;
SSchTask
*
pTask
=
NULL
;
SSchJob
*
pJob
=
schAcquireJob
(
pParam
->
refId
);
if
(
NULL
==
pJob
)
{
...
...
@@ -1617,8 +1617,8 @@ _return:
int32_t
schMakeHbRpcCtx
(
SSchJob
*
pJob
,
SSchTask
*
pTask
,
SRpcCtx
*
pCtx
)
{
int32_t
code
=
0
;
SSchHbCallbackParam
*
param
=
NULL
;
SMsgSendInfo
*
pMsgSendInfo
=
NULL
;
SQueryNodeAddr
*
addr
=
taosArrayGet
(
pTask
->
candidateAddrs
,
pTask
->
candidateIdx
);
SMsgSendInfo
*
pMsgSendInfo
=
NULL
;
SQueryNodeAddr
*
addr
=
taosArrayGet
(
pTask
->
candidateAddrs
,
pTask
->
candidateIdx
);
SQueryNodeEpId
epId
=
{
0
};
epId
.
nodeId
=
addr
->
nodeId
;
...
...
@@ -1759,10 +1759,10 @@ int32_t schCloneHbRpcCtx(SRpcCtx *pSrc, SRpcCtx *pDst) {
}
SRpcCtxVal
dst
=
{
0
};
void
*
pIter
=
taosHashIterate
(
pSrc
->
args
,
NULL
);
void
*
pIter
=
taosHashIterate
(
pSrc
->
args
,
NULL
);
while
(
pIter
)
{
SRpcCtxVal
*
pVal
=
(
SRpcCtxVal
*
)
pIter
;
int32_t
*
msgType
=
taosHashGetKey
(
pIter
,
NULL
);
int32_t
*
msgType
=
taosHashGetKey
(
pIter
,
NULL
);
dst
=
*
pVal
;
dst
.
val
=
NULL
;
...
...
@@ -1916,7 +1916,7 @@ _return:
int32_t
schBuildAndSendMsg
(
SSchJob
*
pJob
,
SSchTask
*
pTask
,
SQueryNodeAddr
*
addr
,
int32_t
msgType
)
{
uint32_t
msgSize
=
0
;
void
*
msg
=
NULL
;
void
*
msg
=
NULL
;
int32_t
code
=
0
;
bool
isCandidateAddr
=
false
;
bool
persistHandle
=
false
;
...
...
@@ -2673,7 +2673,7 @@ int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub) {
SSchLevel
*
pLevel
=
taosArrayGet
(
pJob
->
levels
,
i
);
for
(
int32_t
m
=
0
;
m
<
pLevel
->
taskNum
;
++
m
)
{
SSchTask
*
pTask
=
taosArrayGet
(
pLevel
->
subTasks
,
m
);
SSchTask
*
pTask
=
taosArrayGet
(
pLevel
->
subTasks
,
m
);
SQuerySubDesc
subDesc
=
{.
tid
=
pTask
->
taskId
,
.
status
=
pTask
->
status
};
taosArrayPush
(
pSub
,
&
subDesc
);
...
...
source/libs/sync/src/syncIO.c
浏览文件 @
76acde1e
source/libs/transport/inc/transComm.h
浏览文件 @
76acde1e
...
...
@@ -327,6 +327,10 @@ void transQueueClear(STransQueue* queue);
*/
void
transQueueDestroy
(
STransQueue
*
queue
);
/*
* init global func
*/
void
transThreadOnce
();
#ifdef __cplusplus
}
#endif
...
...
source/libs/transport/src/transComm.c
浏览文件 @
76acde1e
...
...
@@ -16,6 +16,8 @@
#include "transComm.h"
// static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;
int
transAuthenticateMsg
(
void
*
pMsg
,
int
msgLen
,
void
*
pAuth
,
void
*
pKey
)
{
T_MD5_CTX
context
;
int
ret
=
-
1
;
...
...
@@ -361,5 +363,10 @@ void transQueueDestroy(STransQueue* queue) {
transQueueClear
(
queue
);
taosArrayDestroy
(
queue
->
q
);
}
// int32_t transGetExHandle() {
// static
//}
// void transThreadOnce() {
// taosThreadOnce(&transModuleInit, );
//}
#endif
source/libs/transport/src/transSrv.c
浏览文件 @
76acde1e
...
...
@@ -17,6 +17,11 @@
#include "transComm.h"
static
TdThreadOnce
transModuleInit
=
PTHREAD_ONCE_INIT
;
static
char
*
notify
=
"a"
;
static
int
transSrvInst
=
0
;
typedef
struct
{
int
notifyCount
;
//
int
init
;
// init or not
...
...
@@ -45,6 +50,7 @@ typedef struct SSrvConn {
struct
sockaddr_in
addr
;
struct
sockaddr_in
locaddr
;
int64_t
refId
;
int
spi
;
char
info
[
64
];
char
user
[
TSDB_UNI_LEN
];
// user ID for the link
...
...
@@ -89,7 +95,12 @@ typedef struct SServerObj {
uv_async_t
*
pAcceptAsync
;
// just to quit from from accept thread
}
SServerObj
;
static
const
char
*
notify
=
"a"
;
// handle
typedef
struct
SExHandle
{
void
*
handle
;
int64_t
refId
;
SWorkThrdObj
*
pThrd
;
}
SExHandle
;
static
void
uvAllocConnBufferCb
(
uv_handle_t
*
handle
,
size_t
suggested_size
,
uv_buf_t
*
buf
);
static
void
uvAllocRecvBufferCb
(
uv_handle_t
*
handle
,
size_t
suggested_size
,
uv_buf_t
*
buf
);
...
...
@@ -130,6 +141,17 @@ static void uvHandleRegister(SSrvMsg* msg, SWorkThrdObj* thrd);
static
void
(
*
transAsyncHandle
[])(
SSrvMsg
*
msg
,
SWorkThrdObj
*
thrd
)
=
{
uvHandleResp
,
uvHandleQuit
,
uvHandleRelease
,
uvHandleRegister
};
static
int32_t
exHandlesMgt
;
void
uvInitExHandleMgt
();
void
uvOpenExHandleMgt
(
int
size
);
void
uvCloseExHandleMgt
();
int64_t
uvAddExHandle
(
void
*
p
);
int32_t
uvRemoveExHandle
(
int64_t
refId
);
int32_t
uvReleaseExHandle
(
int64_t
refId
);
void
uvDestoryExHandle
(
void
*
handle
);
SExHandle
*
uvAcquireExHandle
(
int64_t
refId
);
static
void
uvDestroyConn
(
uv_handle_t
*
handle
);
// server and worker thread
...
...
@@ -168,6 +190,40 @@ static bool addHandleToAcceptloop(void* arg);
uv_loop_close(loop); \
} while (0);
#define ASYNC_ERR_JRET(thrd) \
do { \
if (thrd->quit) { \
tTrace("worker thread already quit, ignore msg"); \
goto _return1; \
} \
} while (0)
#define ASYNC_CHECK_HANDLE(exh1, refId) \
do { \
if (refId > 0) { \
tTrace("server handle step1"); \
SExHandle* exh2 = uvAcquireExHandle(refId); \
if (exh2 == NULL || refId != exh2->refId) { \
tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
exh1->refId, refId); \
goto _return1; \
} \
} else if (refId == 0) { \
tTrace("server handle step2"); \
SExHandle* exh2 = uvAcquireExHandle(refId); \
if (exh2 == NULL || refId != exh2->refId) { \
tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
refId, exh2 ? exh2->refId : 0); \
goto _return1; \
} else { \
refId = exh1->refId; \
} \
} else if (refId == -1) { \
tTrace("server handle step3"); \
goto _return2; \
} \
} while (0)
void
uvAllocRecvBufferCb
(
uv_handle_t
*
handle
,
size_t
suggested_size
,
uv_buf_t
*
buf
)
{
SSrvConn
*
conn
=
handle
->
data
;
SConnBuffer
*
pBuf
=
&
conn
->
readBuf
;
...
...
@@ -233,7 +289,15 @@ static void uvHandleReq(SSrvConn* pConn) {
// 1. server application should not send resp on handle
// 2. once send out data, cli conn released to conn pool immediately
// 3. not mixed with persist
transMsg
.
handle
=
pConn
;
transMsg
.
handle
=
(
void
*
)
uvAcquireExHandle
(
pConn
->
refId
);
tTrace
(
"server handle %p conn: %p translated to app, refId: %"
PRIu64
""
,
transMsg
.
handle
,
pConn
,
pConn
->
refId
);
transMsg
.
refId
=
pConn
->
refId
;
assert
(
transMsg
.
handle
!=
NULL
);
if
(
pHead
->
noResp
==
1
)
{
transMsg
.
refId
=
-
1
;
}
uvReleaseExHandle
(
pConn
->
refId
);
STrans
*
pTransInst
=
pConn
->
pTransInst
;
(
*
pTransInst
->
cfp
)(
pTransInst
->
parent
,
&
transMsg
,
NULL
);
...
...
@@ -404,6 +468,7 @@ static void destroySmsg(SSrvMsg* smsg) {
taosMemoryFree
(
smsg
);
}
static
void
destroyAllConn
(
SWorkThrdObj
*
pThrd
)
{
tTrace
(
"thread %p destroy all conn "
,
pThrd
);
while
(
!
QUEUE_IS_EMPTY
(
&
pThrd
->
conn
))
{
queue
*
h
=
QUEUE_HEAD
(
&
pThrd
->
conn
);
QUEUE_REMOVE
(
h
);
...
...
@@ -436,7 +501,26 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
tError
(
"unexcept occurred, continue"
);
continue
;
}
// release handle to rpc init
if
(
msg
->
type
==
Quit
)
{
(
*
transAsyncHandle
[
msg
->
type
])(
msg
,
pThrd
);
continue
;
}
else
{
STransMsg
transMsg
=
msg
->
msg
;
SExHandle
*
exh1
=
transMsg
.
handle
;
int64_t
refId
=
transMsg
.
refId
;
SExHandle
*
exh2
=
uvAcquireExHandle
(
refId
);
if
(
exh2
==
NULL
||
exh1
!=
exh2
)
{
tTrace
(
"server handle %p except msg, ignore it"
,
exh1
);
uvReleaseExHandle
(
refId
);
destroySmsg
(
msg
);
continue
;
}
msg
->
pConn
=
exh1
->
handle
;
uvReleaseExHandle
(
refId
);
(
*
transAsyncHandle
[
msg
->
type
])(
msg
,
pThrd
);
}
}
}
static
void
uvWalkCb
(
uv_handle_t
*
handle
,
void
*
arg
)
{
...
...
@@ -658,8 +742,15 @@ static SSrvConn* createConn(void* hThrd) {
pConn
->
broken
=
false
;
pConn
->
status
=
ConnNormal
;
SExHandle
*
exh
=
taosMemoryMalloc
(
sizeof
(
SExHandle
));
exh
->
handle
=
pConn
;
exh
->
pThrd
=
pThrd
;
exh
->
refId
=
uvAddExHandle
(
exh
);
uvAcquireExHandle
(
exh
->
refId
);
pConn
->
refId
=
exh
->
refId
;
transRefSrvHandle
(
pConn
);
tTrace
(
"server
conn %p created"
,
pConn
);
tTrace
(
"server
handle %p, conn %p created, refId: %"
PRId64
""
,
exh
,
pConn
,
pConn
->
refId
);
return
pConn
;
}
...
...
@@ -667,6 +758,7 @@ static void destroyConn(SSrvConn* conn, bool clear) {
if
(
conn
==
NULL
)
{
return
;
}
transDestroyBuffer
(
&
conn
->
readBuf
);
if
(
clear
)
{
tTrace
(
"server conn %p to be destroyed"
,
conn
);
...
...
@@ -681,6 +773,9 @@ static void uvDestroyConn(uv_handle_t* handle) {
}
SWorkThrdObj
*
thrd
=
conn
->
hostThrd
;
uvReleaseExHandle
(
conn
->
refId
);
uvRemoveExHandle
(
conn
->
refId
);
tDebug
(
"server conn %p destroy"
,
conn
);
// uv_timer_stop(&conn->pTimer);
transQueueDestroy
(
&
conn
->
srvMsgs
);
...
...
@@ -705,6 +800,10 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
srv
->
port
=
port
;
uv_loop_init
(
srv
->
loop
);
taosThreadOnce
(
&
transModuleInit
,
uvInitExHandleMgt
);
transSrvInst
++
;
// uvOpenExHandleMgt(10000);
for
(
int
i
=
0
;
i
<
srv
->
numOfThreads
;
i
++
)
{
SWorkThrdObj
*
thrd
=
(
SWorkThrdObj
*
)
taosMemoryCalloc
(
1
,
sizeof
(
SWorkThrdObj
));
thrd
->
quit
=
false
;
...
...
@@ -749,6 +848,44 @@ End:
transCloseServer
(
srv
);
return
NULL
;
}
void
uvInitExHandleMgt
()
{
// init exhandle mgt
uvOpenExHandleMgt
(
10000
);
}
void
uvOpenExHandleMgt
(
int
size
)
{
// added into once later
exHandlesMgt
=
taosOpenRef
(
size
,
uvDestoryExHandle
);
}
void
uvCloseExHandleMgt
()
{
// close ref
taosCloseRef
(
exHandlesMgt
);
}
int64_t
uvAddExHandle
(
void
*
p
)
{
// acquire extern handle
return
taosAddRef
(
exHandlesMgt
,
p
);
}
int32_t
uvRemoveExHandle
(
int64_t
refId
)
{
// acquire extern handle
return
taosRemoveRef
(
exHandlesMgt
,
refId
);
}
SExHandle
*
uvAcquireExHandle
(
int64_t
refId
)
{
// acquire extern handle
return
(
SExHandle
*
)
taosAcquireRef
(
exHandlesMgt
,
refId
);
}
int32_t
uvReleaseExHandle
(
int64_t
refId
)
{
// release extern handle
return
taosReleaseRef
(
exHandlesMgt
,
refId
);
}
void
uvDestoryExHandle
(
void
*
handle
)
{
if
(
handle
==
NULL
)
{
return
;
}
taosMemoryFree
(
handle
);
}
void
uvHandleQuit
(
SSrvMsg
*
msg
,
SWorkThrdObj
*
thrd
)
{
thrd
->
quit
=
true
;
if
(
QUEUE_IS_EMPTY
(
&
thrd
->
conn
))
{
...
...
@@ -759,7 +896,6 @@ void uvHandleQuit(SSrvMsg* msg, SWorkThrdObj* thrd) {
taosMemoryFree
(
msg
);
}
void
uvHandleRelease
(
SSrvMsg
*
msg
,
SWorkThrdObj
*
thrd
)
{
// release handle to rpc init
SSrvConn
*
conn
=
msg
->
pConn
;
if
(
conn
->
status
==
ConnAcquire
)
{
if
(
!
transQueuePush
(
&
conn
->
srvMsgs
,
msg
))
{
...
...
@@ -840,6 +976,12 @@ void transCloseServer(void* arg) {
taosMemoryFree
(
srv
->
pipe
);
taosMemoryFree
(
srv
);
transSrvInst
--
;
if
(
transSrvInst
==
0
)
{
transModuleInit
=
PTHREAD_ONCE_INIT
;
uvCloseExHandleMgt
();
}
}
void
transRefSrvHandle
(
void
*
handle
)
{
...
...
@@ -862,57 +1004,98 @@ void transUnrefSrvHandle(void* handle) {
}
void
transReleaseSrvHandle
(
void
*
handle
)
{
if
(
handle
==
NULL
)
{
return
;
}
SSrvConn
*
pConn
=
handle
;
SWorkThrdObj
*
pThrd
=
pConn
->
hostThrd
;
SExHandle
*
exh
=
handle
;
int64_t
refId
=
exh
->
refId
;
ASYNC_CHECK_HANDLE
(
exh
,
refId
);
SWorkThrdObj
*
pThrd
=
exh
->
pThrd
;
ASYNC_ERR_JRET
(
pThrd
);
STransMsg
tmsg
=
{.
code
=
0
,
.
handle
=
handle
,
.
ahandle
=
NULL
};
STransMsg
tmsg
=
{.
code
=
0
,
.
handle
=
exh
,
.
ahandle
=
NULL
,
.
refId
=
refId
};
SSrvMsg
*
srvMsg
=
taosMemoryCalloc
(
1
,
sizeof
(
SSrvMsg
));
srvMsg
->
msg
=
tmsg
;
srvMsg
->
type
=
Release
;
srvMsg
->
pConn
=
pConn
;
tTrace
(
"server conn %p start to release"
,
pConn
);
tTrace
(
"server conn %p start to release"
,
exh
->
handle
);
transSendAsync
(
pThrd
->
asyncPool
,
&
srvMsg
->
q
);
}
void
transSendResponse
(
const
STransMsg
*
pMsg
)
{
if
(
pMsg
->
handle
==
NULL
)
{
uvReleaseExHandle
(
refId
);
return
;
}
SSrvConn
*
pConn
=
pMsg
->
handle
;
SWorkThrdObj
*
pThrd
=
pConn
->
hostThrd
;
if
(
pThrd
->
quit
)
{
_return1:
tTrace
(
"server handle %p failed to send to release handle"
,
exh
);
uvReleaseExHandle
(
refId
);
return
;
}
_return2:
tTrace
(
"server handle %p failed to send to release handle"
,
exh
);
return
;
}
void
transSendResponse
(
const
STransMsg
*
msg
)
{
SExHandle
*
exh
=
msg
->
handle
;
int64_t
refId
=
msg
->
refId
;
ASYNC_CHECK_HANDLE
(
exh
,
refId
);
assert
(
refId
!=
0
);
STransMsg
tmsg
=
*
msg
;
tmsg
.
refId
=
refId
;
SWorkThrdObj
*
pThrd
=
exh
->
pThrd
;
ASYNC_ERR_JRET
(
pThrd
);
SSrvMsg
*
srvMsg
=
taosMemoryCalloc
(
1
,
sizeof
(
SSrvMsg
));
srvMsg
->
pConn
=
pConn
;
srvMsg
->
msg
=
*
pMsg
;
srvMsg
->
msg
=
tmsg
;
srvMsg
->
type
=
Normal
;
tTrace
(
"server conn %p start to send resp (1/2)"
,
pConn
);
tTrace
(
"server conn %p start to send resp (1/2)"
,
exh
->
handle
);
transSendAsync
(
pThrd
->
asyncPool
,
&
srvMsg
->
q
);
uvReleaseExHandle
(
refId
);
return
;
_return1:
tTrace
(
"server handle %p failed to send resp"
,
exh
);
rpcFreeCont
(
msg
->
pCont
);
uvReleaseExHandle
(
refId
);
return
;
_return2:
tTrace
(
"server handle %p failed to send resp"
,
exh
);
rpcFreeCont
(
msg
->
pCont
);
return
;
}
void
transRegisterMsg
(
const
STransMsg
*
msg
)
{
if
(
msg
->
handle
==
NULL
)
{
return
;
}
SSrvConn
*
pConn
=
msg
->
handle
;
SWorkThrdObj
*
pThrd
=
pConn
->
hostThrd
;
SExHandle
*
exh
=
msg
->
handle
;
int64_t
refId
=
msg
->
refId
;
ASYNC_CHECK_HANDLE
(
exh
,
refId
);
STransMsg
tmsg
=
*
msg
;
tmsg
.
refId
=
refId
;
SWorkThrdObj
*
pThrd
=
exh
->
pThrd
;
ASYNC_ERR_JRET
(
pThrd
);
SSrvMsg
*
srvMsg
=
taosMemoryCalloc
(
1
,
sizeof
(
SSrvMsg
));
srvMsg
->
pConn
=
pConn
;
srvMsg
->
msg
=
*
msg
;
srvMsg
->
msg
=
tmsg
;
srvMsg
->
type
=
Register
;
tTrace
(
"server conn %p start to register brokenlink callback"
,
pConn
);
tTrace
(
"server conn %p start to register brokenlink callback"
,
exh
->
handle
);
transSendAsync
(
pThrd
->
asyncPool
,
&
srvMsg
->
q
);
uvReleaseExHandle
(
refId
);
return
;
_return1:
tTrace
(
"server handle %p failed to send to register brokenlink"
,
exh
);
rpcFreeCont
(
msg
->
pCont
);
uvReleaseExHandle
(
refId
);
return
;
_return2:
tTrace
(
"server handle %p failed to send to register brokenlink"
,
exh
);
rpcFreeCont
(
msg
->
pCont
);
}
int
transGetConnInfo
(
void
*
thandle
,
STransHandleInfo
*
pInfo
)
{
SSrvConn
*
pConn
=
thandle
;
struct
sockaddr_in
addr
=
pConn
->
addr
;
if
(
thandle
==
NULL
)
{
tTrace
(
"invalid handle %p, failed to Get Conn info"
,
thandle
);
return
-
1
;
}
SExHandle
*
ex
=
thandle
;
SSrvConn
*
pConn
=
ex
->
handle
;
struct
sockaddr_in
addr
=
pConn
->
addr
;
pInfo
->
clientIp
=
(
uint32_t
)(
addr
.
sin_addr
.
s_addr
);
pInfo
->
clientPort
=
ntohs
(
addr
.
sin_port
);
tstrncpy
(
pInfo
->
user
,
pConn
->
user
,
sizeof
(
pInfo
->
user
));
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录