Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
997a00ce
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
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看板
提交
997a00ce
编写于
11月 23, 2022
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
差异文件
Merge branch '3.0' into fix/TD-20631
上级
10333bdd
188e6e6c
变更
8
显示空白变更内容
内联
并排
Showing
8 changed file
with
252 addition
and
76 deletion
+252
-76
include/common/tmsg.h
include/common/tmsg.h
+8
-1
source/client/src/clientTmq.c
source/client/src/clientTmq.c
+27
-17
source/common/src/tmsg.c
source/common/src/tmsg.c
+133
-0
source/dnode/vnode/src/tq/tq.c
source/dnode/vnode/src/tq/tq.c
+31
-25
source/libs/executor/src/exchangeoperator.c
source/libs/executor/src/exchangeoperator.c
+26
-11
source/libs/qworker/src/qwMsg.c
source/libs/qworker/src/qwMsg.c
+7
-12
source/libs/scheduler/src/schRemote.c
source/libs/scheduler/src/schRemote.c
+17
-10
tests/script/runAllSimCases.sh
tests/script/runAllSimCases.sh
+3
-0
未找到文件。
include/common/tmsg.h
浏览文件 @
997a00ce
...
@@ -1629,7 +1629,6 @@ int32_t tSerializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *pReq);
...
@@ -1629,7 +1629,6 @@ int32_t tSerializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *pReq);
int32_t
tDeserializeSSubQueryMsg
(
void
*
buf
,
int32_t
bufLen
,
SSubQueryMsg
*
pReq
);
int32_t
tDeserializeSSubQueryMsg
(
void
*
buf
,
int32_t
bufLen
,
SSubQueryMsg
*
pReq
);
void
tFreeSSubQueryMsg
(
SSubQueryMsg
*
pReq
);
void
tFreeSSubQueryMsg
(
SSubQueryMsg
*
pReq
);
typedef
struct
{
typedef
struct
{
SMsgHead
header
;
SMsgHead
header
;
uint64_t
sId
;
uint64_t
sId
;
...
@@ -1667,6 +1666,10 @@ typedef struct {
...
@@ -1667,6 +1666,10 @@ typedef struct {
int32_t
execId
;
int32_t
execId
;
}
SResFetchReq
;
}
SResFetchReq
;
int32_t
tSerializeSResFetchReq
(
void
*
buf
,
int32_t
bufLen
,
SResFetchReq
*
pReq
);
int32_t
tDeserializeSResFetchReq
(
void
*
buf
,
int32_t
bufLen
,
SResFetchReq
*
pReq
);
typedef
struct
{
typedef
struct
{
SMsgHead
header
;
SMsgHead
header
;
uint64_t
sId
;
uint64_t
sId
;
...
@@ -2948,6 +2951,10 @@ typedef struct {
...
@@ -2948,6 +2951,10 @@ typedef struct {
STqOffsetVal
reqOffset
;
STqOffsetVal
reqOffset
;
}
SMqPollReq
;
}
SMqPollReq
;
int32_t
tSerializeSMqPollReq
(
void
*
buf
,
int32_t
bufLen
,
SMqPollReq
*
pReq
);
int32_t
tDeserializeSMqPollReq
(
void
*
buf
,
int32_t
bufLen
,
SMqPollReq
*
pReq
);
typedef
struct
{
typedef
struct
{
int32_t
vgId
;
int32_t
vgId
;
int64_t
offset
;
int64_t
offset
;
...
...
source/client/src/clientTmq.c
浏览文件 @
997a00ce
...
@@ -1461,12 +1461,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool async) {
...
@@ -1461,12 +1461,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool async) {
return
code
;
return
code
;
}
}
SMqPollReq
*
tmqBuildConsumeReqImpl
(
tmq_t
*
tmq
,
int64_t
timeout
,
SMqClientTopic
*
pTopic
,
SMqClientVg
*
pVg
)
{
void
tmqBuildConsumeReqImpl
(
SMqPollReq
*
pReq
,
tmq_t
*
tmq
,
int64_t
timeout
,
SMqClientTopic
*
pTopic
,
SMqClientVg
*
pVg
)
{
SMqPollReq
*
pReq
=
taosMemoryCalloc
(
1
,
sizeof
(
SMqPollReq
));
if
(
pReq
==
NULL
)
{
return
NULL
;
}
/*strcpy(pReq->topic, pTopic->topicName);*/
/*strcpy(pReq->topic, pTopic->topicName);*/
/*strcpy(pReq->cgroup, tmq->groupId);*/
/*strcpy(pReq->cgroup, tmq->groupId);*/
...
@@ -1485,9 +1480,7 @@ SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t timeout, SMqClientTopic*
...
@@ -1485,9 +1480,7 @@ SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t timeout, SMqClientTopic*
pReq
->
useSnapshot
=
tmq
->
useSnapshot
;
pReq
->
useSnapshot
=
tmq
->
useSnapshot
;
pReq
->
head
.
vgId
=
htonl
(
pVg
->
vgId
);
pReq
->
head
.
vgId
=
pVg
->
vgId
;
pReq
->
head
.
contLen
=
htonl
(
sizeof
(
SMqPollReq
));
return
pReq
;
}
}
SMqMetaRspObj
*
tmqBuildMetaRspFromWrapper
(
SMqPollRspWrapper
*
pWrapper
)
{
SMqMetaRspObj
*
tmqBuildMetaRspFromWrapper
(
SMqPollRspWrapper
*
pWrapper
)
{
...
@@ -1559,15 +1552,32 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
...
@@ -1559,15 +1552,32 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
#endif
#endif
}
}
atomic_store_32
(
&
pVg
->
vgSkipCnt
,
0
);
atomic_store_32
(
&
pVg
->
vgSkipCnt
,
0
);
SMqPollReq
*
pReq
=
tmqBuildConsumeReqImpl
(
tmq
,
timeout
,
pTopic
,
pVg
);
if
(
pReq
==
NULL
)
{
SMqPollReq
req
=
{
0
};
tmqBuildConsumeReqImpl
(
&
req
,
tmq
,
timeout
,
pTopic
,
pVg
);
int32_t
msgSize
=
tSerializeSMqPollReq
(
NULL
,
0
,
&
req
);
if
(
msgSize
<
0
)
{
atomic_store_32
(
&
pVg
->
vgStatus
,
TMQ_VG_STATUS__IDLE
);
atomic_store_32
(
&
pVg
->
vgStatus
,
TMQ_VG_STATUS__IDLE
);
tsem_post
(
&
tmq
->
rspSem
);
tsem_post
(
&
tmq
->
rspSem
);
return
-
1
;
return
-
1
;
}
}
char
*
msg
=
taosMemoryCalloc
(
1
,
msgSize
);
if
(
NULL
==
msg
)
{
atomic_store_32
(
&
pVg
->
vgStatus
,
TMQ_VG_STATUS__IDLE
);
tsem_post
(
&
tmq
->
rspSem
);
return
-
1
;
}
if
(
tSerializeSMqPollReq
(
msg
,
msgSize
,
&
req
)
<
0
)
{
taosMemoryFree
(
msg
);
atomic_store_32
(
&
pVg
->
vgStatus
,
TMQ_VG_STATUS__IDLE
);
tsem_post
(
&
tmq
->
rspSem
);
return
-
1
;
}
SMqPollCbParam
*
pParam
=
taosMemoryMalloc
(
sizeof
(
SMqPollCbParam
));
SMqPollCbParam
*
pParam
=
taosMemoryMalloc
(
sizeof
(
SMqPollCbParam
));
if
(
pParam
==
NULL
)
{
if
(
pParam
==
NULL
)
{
taosMemoryFree
(
pReq
);
taosMemoryFree
(
msg
);
atomic_store_32
(
&
pVg
->
vgStatus
,
TMQ_VG_STATUS__IDLE
);
atomic_store_32
(
&
pVg
->
vgStatus
,
TMQ_VG_STATUS__IDLE
);
tsem_post
(
&
tmq
->
rspSem
);
tsem_post
(
&
tmq
->
rspSem
);
return
-
1
;
return
-
1
;
...
@@ -1581,7 +1591,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
...
@@ -1581,7 +1591,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
SMsgSendInfo
*
sendInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SMsgSendInfo
));
SMsgSendInfo
*
sendInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SMsgSendInfo
));
if
(
sendInfo
==
NULL
)
{
if
(
sendInfo
==
NULL
)
{
taosMemoryFree
(
pReq
);
taosMemoryFree
(
msg
);
taosMemoryFree
(
pParam
);
taosMemoryFree
(
pParam
);
atomic_store_32
(
&
pVg
->
vgStatus
,
TMQ_VG_STATUS__IDLE
);
atomic_store_32
(
&
pVg
->
vgStatus
,
TMQ_VG_STATUS__IDLE
);
tsem_post
(
&
tmq
->
rspSem
);
tsem_post
(
&
tmq
->
rspSem
);
...
@@ -1589,11 +1599,11 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
...
@@ -1589,11 +1599,11 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
}
}
sendInfo
->
msgInfo
=
(
SDataBuf
){
sendInfo
->
msgInfo
=
(
SDataBuf
){
.
pData
=
pReq
,
.
pData
=
msg
,
.
len
=
sizeof
(
SMqPollReq
)
,
.
len
=
msgSize
,
.
handle
=
NULL
,
.
handle
=
NULL
,
};
};
sendInfo
->
requestId
=
pReq
->
reqId
;
sendInfo
->
requestId
=
req
.
reqId
;
sendInfo
->
requestObjRefId
=
0
;
sendInfo
->
requestObjRefId
=
0
;
sendInfo
->
param
=
pParam
;
sendInfo
->
param
=
pParam
;
sendInfo
->
fp
=
tmqPollCb
;
sendInfo
->
fp
=
tmqPollCb
;
...
@@ -1605,7 +1615,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
...
@@ -1605,7 +1615,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
char
offsetFormatBuf
[
80
];
char
offsetFormatBuf
[
80
];
tFormatOffset
(
offsetFormatBuf
,
80
,
&
pVg
->
currentOffset
);
tFormatOffset
(
offsetFormatBuf
,
80
,
&
pVg
->
currentOffset
);
tscDebug
(
"consumer:%"
PRId64
", send poll to %s vgId:%d, epoch %d, req offset:%s, reqId:%"
PRIu64
,
tscDebug
(
"consumer:%"
PRId64
", send poll to %s vgId:%d, epoch %d, req offset:%s, reqId:%"
PRIu64
,
tmq
->
consumerId
,
pTopic
->
topicName
,
pVg
->
vgId
,
tmq
->
epoch
,
offsetFormatBuf
,
pReq
->
reqId
);
tmq
->
consumerId
,
pTopic
->
topicName
,
pVg
->
vgId
,
tmq
->
epoch
,
offsetFormatBuf
,
req
.
reqId
);
/*printf("send vgId:%d %" PRId64 "\n", pVg->vgId, pVg->currentOffset);*/
/*printf("send vgId:%d %" PRId64 "\n", pVg->vgId, pVg->currentOffset);*/
asyncSendMsgToServer
(
tmq
->
pTscObj
->
pAppInfo
->
pTransporter
,
&
pVg
->
epSet
,
&
transporterId
,
sendInfo
);
asyncSendMsgToServer
(
tmq
->
pTscObj
->
pAppInfo
->
pTransporter
,
&
pVg
->
epSet
,
&
transporterId
,
sendInfo
);
pVg
->
pollCnt
++
;
pVg
->
pollCnt
++
;
...
...
source/common/src/tmsg.c
浏览文件 @
997a00ce
...
@@ -4723,6 +4723,139 @@ void tFreeSSubQueryMsg(SSubQueryMsg *pReq) {
...
@@ -4723,6 +4723,139 @@ void tFreeSSubQueryMsg(SSubQueryMsg *pReq) {
taosMemoryFreeClear
(
pReq
->
msg
);
taosMemoryFreeClear
(
pReq
->
msg
);
}
}
int32_t
tSerializeSResFetchReq
(
void
*
buf
,
int32_t
bufLen
,
SResFetchReq
*
pReq
)
{
int32_t
headLen
=
sizeof
(
SMsgHead
);
if
(
buf
!=
NULL
)
{
buf
=
(
char
*
)
buf
+
headLen
;
bufLen
-=
headLen
;
}
SEncoder
encoder
=
{
0
};
tEncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeU64
(
&
encoder
,
pReq
->
sId
)
<
0
)
return
-
1
;
if
(
tEncodeU64
(
&
encoder
,
pReq
->
queryId
)
<
0
)
return
-
1
;
if
(
tEncodeU64
(
&
encoder
,
pReq
->
taskId
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
&
encoder
,
pReq
->
execId
)
<
0
)
return
-
1
;
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
tEncoderClear
(
&
encoder
);
if
(
buf
!=
NULL
)
{
SMsgHead
*
pHead
=
(
SMsgHead
*
)((
char
*
)
buf
-
headLen
);
pHead
->
vgId
=
htonl
(
pReq
->
header
.
vgId
);
pHead
->
contLen
=
htonl
(
tlen
+
headLen
);
}
return
tlen
+
headLen
;
}
int32_t
tDeserializeSResFetchReq
(
void
*
buf
,
int32_t
bufLen
,
SResFetchReq
*
pReq
)
{
int32_t
headLen
=
sizeof
(
SMsgHead
);
SMsgHead
*
pHead
=
buf
;
pHead
->
vgId
=
pReq
->
header
.
vgId
;
pHead
->
contLen
=
pReq
->
header
.
contLen
;
SDecoder
decoder
=
{
0
};
tDecoderInit
(
&
decoder
,
(
char
*
)
buf
+
headLen
,
bufLen
-
headLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tDecodeU64
(
&
decoder
,
&
pReq
->
sId
)
<
0
)
return
-
1
;
if
(
tDecodeU64
(
&
decoder
,
&
pReq
->
queryId
)
<
0
)
return
-
1
;
if
(
tDecodeU64
(
&
decoder
,
&
pReq
->
taskId
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
&
decoder
,
&
pReq
->
execId
)
<
0
)
return
-
1
;
tEndDecode
(
&
decoder
);
tDecoderClear
(
&
decoder
);
return
0
;
}
int32_t
tSerializeSTqOffsetVal
(
SEncoder
*
pEncoder
,
STqOffsetVal
*
pOffset
)
{
if
(
tEncodeI8
(
pEncoder
,
pOffset
->
type
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
pEncoder
,
pOffset
->
uid
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
pEncoder
,
pOffset
->
ts
)
<
0
)
return
-
1
;
return
0
;
}
int32_t
tDerializeSTqOffsetVal
(
SDecoder
*
pDecoder
,
STqOffsetVal
*
pOffset
)
{
if
(
tDecodeI8
(
pDecoder
,
&
pOffset
->
type
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
pDecoder
,
&
pOffset
->
uid
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
pDecoder
,
&
pOffset
->
ts
)
<
0
)
return
-
1
;
return
0
;
}
int32_t
tSerializeSMqPollReq
(
void
*
buf
,
int32_t
bufLen
,
SMqPollReq
*
pReq
)
{
int32_t
headLen
=
sizeof
(
SMsgHead
);
if
(
buf
!=
NULL
)
{
buf
=
(
char
*
)
buf
+
headLen
;
bufLen
-=
headLen
;
}
SEncoder
encoder
=
{
0
};
tEncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeCStr
(
&
encoder
,
pReq
->
subKey
)
<
0
)
return
-
1
;
if
(
tEncodeI8
(
&
encoder
,
pReq
->
withTbName
)
<
0
)
return
-
1
;
if
(
tEncodeI8
(
&
encoder
,
pReq
->
useSnapshot
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
&
encoder
,
pReq
->
epoch
)
<
0
)
return
-
1
;
if
(
tEncodeU64
(
&
encoder
,
pReq
->
reqId
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
&
encoder
,
pReq
->
consumerId
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
&
encoder
,
pReq
->
timeout
)
<
0
)
return
-
1
;
if
(
tSerializeSTqOffsetVal
(
&
encoder
,
&
pReq
->
reqOffset
)
<
0
)
return
-
1
;
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
tEncoderClear
(
&
encoder
);
if
(
buf
!=
NULL
)
{
SMsgHead
*
pHead
=
(
SMsgHead
*
)((
char
*
)
buf
-
headLen
);
pHead
->
vgId
=
htonl
(
pReq
->
head
.
vgId
);
pHead
->
contLen
=
htonl
(
tlen
+
headLen
);
}
return
tlen
+
headLen
;
}
int32_t
tDeserializeSMqPollReq
(
void
*
buf
,
int32_t
bufLen
,
SMqPollReq
*
pReq
)
{
int32_t
headLen
=
sizeof
(
SMsgHead
);
SMsgHead
*
pHead
=
buf
;
pHead
->
vgId
=
pReq
->
head
.
vgId
;
pHead
->
contLen
=
pReq
->
head
.
contLen
;
SDecoder
decoder
=
{
0
};
tDecoderInit
(
&
decoder
,
(
char
*
)
buf
+
headLen
,
bufLen
-
headLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tDecodeCStrTo
(
&
decoder
,
pReq
->
subKey
)
<
0
)
return
-
1
;
if
(
tDecodeI8
(
&
decoder
,
&
pReq
->
withTbName
)
<
0
)
return
-
1
;
if
(
tDecodeI8
(
&
decoder
,
&
pReq
->
useSnapshot
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
&
decoder
,
&
pReq
->
epoch
)
<
0
)
return
-
1
;
if
(
tDecodeU64
(
&
decoder
,
&
pReq
->
reqId
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
&
decoder
,
&
pReq
->
consumerId
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
&
decoder
,
&
pReq
->
timeout
)
<
0
)
return
-
1
;
if
(
tDerializeSTqOffsetVal
(
&
decoder
,
&
pReq
->
reqOffset
)
<
0
)
return
-
1
;
tEndDecode
(
&
decoder
);
tDecoderClear
(
&
decoder
);
return
0
;
}
int32_t
tSerializeSTaskDropReq
(
void
*
buf
,
int32_t
bufLen
,
STaskDropReq
*
pReq
)
{
int32_t
tSerializeSTaskDropReq
(
void
*
buf
,
int32_t
bufLen
,
STaskDropReq
*
pReq
)
{
int32_t
headLen
=
sizeof
(
SMsgHead
);
int32_t
headLen
=
sizeof
(
SMsgHead
);
if
(
buf
!=
NULL
)
{
if
(
buf
!=
NULL
)
{
...
...
source/dnode/vnode/src/tq/tq.c
浏览文件 @
997a00ce
...
@@ -458,20 +458,26 @@ static int32_t tqInitTaosxRsp(STaosxRsp* pRsp, const SMqPollReq* pReq) {
...
@@ -458,20 +458,26 @@ static int32_t tqInitTaosxRsp(STaosxRsp* pRsp, const SMqPollReq* pReq) {
}
}
int32_t
tqProcessPollReq
(
STQ
*
pTq
,
SRpcMsg
*
pMsg
)
{
int32_t
tqProcessPollReq
(
STQ
*
pTq
,
SRpcMsg
*
pMsg
)
{
SMqPollReq
*
pReq
=
pMsg
->
pCont
;
SMqPollReq
req
=
{
0
};
int64_t
consumerId
=
pReq
->
consumerId
;
int32_t
reqEpoch
=
pReq
->
epoch
;
int32_t
code
=
0
;
int32_t
code
=
0
;
STqOffsetVal
reqOffset
=
pReq
->
reqOffset
;
STqOffsetVal
fetchOffsetNew
;
STqOffsetVal
fetchOffsetNew
;
SWalCkHead
*
pCkHead
=
NULL
;
SWalCkHead
*
pCkHead
=
NULL
;
if
(
tDeserializeSMqPollReq
(
pMsg
->
pCont
,
pMsg
->
contLen
,
&
req
)
<
0
)
{
tqError
(
"tDeserializeSMqPollReq %d failed"
,
pMsg
->
contLen
);
return
-
1
;
}
int64_t
consumerId
=
req
.
consumerId
;
int32_t
reqEpoch
=
req
.
epoch
;
STqOffsetVal
reqOffset
=
req
.
reqOffset
;
// 1.find handle
// 1.find handle
STqHandle
*
pHandle
=
taosHashGet
(
pTq
->
pHandle
,
pReq
->
subKey
,
strlen
(
pReq
->
subKey
));
STqHandle
*
pHandle
=
taosHashGet
(
pTq
->
pHandle
,
req
.
subKey
,
strlen
(
req
.
subKey
));
/*ASSERT(pHandle);*/
/*ASSERT(pHandle);*/
if
(
pHandle
==
NULL
)
{
if
(
pHandle
==
NULL
)
{
tqError
(
"tmq poll: no consumer handle for consumer:%"
PRId64
", in vgId:%d, subkey %s"
,
consumerId
,
tqError
(
"tmq poll: no consumer handle for consumer:%"
PRId64
", in vgId:%d, subkey %s"
,
consumerId
,
TD_VID
(
pTq
->
pVnode
),
pReq
->
subKey
);
TD_VID
(
pTq
->
pVnode
),
req
.
subKey
);
return
-
1
;
return
-
1
;
}
}
...
@@ -479,7 +485,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -479,7 +485,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
if
(
pHandle
->
consumerId
!=
consumerId
)
{
if
(
pHandle
->
consumerId
!=
consumerId
)
{
tqError
(
"tmq poll: consumer handle mismatch for consumer:%"
PRId64
tqError
(
"tmq poll: consumer handle mismatch for consumer:%"
PRId64
", in vgId:%d, subkey %s, handle consumer id %"
PRId64
,
", in vgId:%d, subkey %s, handle consumer id %"
PRId64
,
consumerId
,
TD_VID
(
pTq
->
pVnode
),
pReq
->
subKey
,
pHandle
->
consumerId
);
consumerId
,
TD_VID
(
pTq
->
pVnode
),
req
.
subKey
,
pHandle
->
consumerId
);
terrno
=
TSDB_CODE_TMQ_CONSUMER_MISMATCH
;
terrno
=
TSDB_CODE_TMQ_CONSUMER_MISMATCH
;
return
-
1
;
return
-
1
;
}
}
...
@@ -493,13 +499,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -493,13 +499,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
char
buf
[
80
];
char
buf
[
80
];
tFormatOffset
(
buf
,
80
,
&
reqOffset
);
tFormatOffset
(
buf
,
80
,
&
reqOffset
);
tqDebug
(
"tmq poll: consumer %"
PRId64
" (epoch %d), subkey %s, recv poll req in vg %d, req offset %s"
,
consumerId
,
tqDebug
(
"tmq poll: consumer %"
PRId64
" (epoch %d), subkey %s, recv poll req in vg %d, req offset %s"
,
consumerId
,
pReq
->
epoch
,
pHandle
->
subKey
,
TD_VID
(
pTq
->
pVnode
),
buf
);
req
.
epoch
,
pHandle
->
subKey
,
TD_VID
(
pTq
->
pVnode
),
buf
);
// 2.reset offset if needed
// 2.reset offset if needed
if
(
reqOffset
.
type
>
0
)
{
if
(
reqOffset
.
type
>
0
)
{
fetchOffsetNew
=
reqOffset
;
fetchOffsetNew
=
reqOffset
;
}
else
{
}
else
{
STqOffset
*
pOffset
=
tqOffsetRead
(
pTq
->
pOffsetStore
,
pReq
->
subKey
);
STqOffset
*
pOffset
=
tqOffsetRead
(
pTq
->
pOffsetStore
,
req
.
subKey
);
if
(
pOffset
!=
NULL
)
{
if
(
pOffset
!=
NULL
)
{
fetchOffsetNew
=
pOffset
->
val
;
fetchOffsetNew
=
pOffset
->
val
;
char
formatBuf
[
80
];
char
formatBuf
[
80
];
...
@@ -508,7 +514,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -508,7 +514,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
TD_VID
(
pTq
->
pVnode
),
formatBuf
);
TD_VID
(
pTq
->
pVnode
),
formatBuf
);
}
else
{
}
else
{
if
(
reqOffset
.
type
==
TMQ_OFFSET__RESET_EARLIEAST
)
{
if
(
reqOffset
.
type
==
TMQ_OFFSET__RESET_EARLIEAST
)
{
if
(
pReq
->
useSnapshot
)
{
if
(
req
.
useSnapshot
)
{
if
(
pHandle
->
fetchMeta
)
{
if
(
pHandle
->
fetchMeta
)
{
tqOffsetResetToMeta
(
&
fetchOffsetNew
,
0
);
tqOffsetResetToMeta
(
&
fetchOffsetNew
,
0
);
}
else
{
}
else
{
...
@@ -520,21 +526,21 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -520,21 +526,21 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
}
else
if
(
reqOffset
.
type
==
TMQ_OFFSET__RESET_LATEST
)
{
}
else
if
(
reqOffset
.
type
==
TMQ_OFFSET__RESET_LATEST
)
{
if
(
pHandle
->
execHandle
.
subType
==
TOPIC_SUB_TYPE__COLUMN
)
{
if
(
pHandle
->
execHandle
.
subType
==
TOPIC_SUB_TYPE__COLUMN
)
{
SMqDataRsp
dataRsp
=
{
0
};
SMqDataRsp
dataRsp
=
{
0
};
tqInitDataRsp
(
&
dataRsp
,
pR
eq
,
pHandle
->
execHandle
.
subType
);
tqInitDataRsp
(
&
dataRsp
,
&
r
eq
,
pHandle
->
execHandle
.
subType
);
tqOffsetResetToLog
(
&
dataRsp
.
rspOffset
,
walGetLastVer
(
pTq
->
pVnode
->
pWal
));
tqOffsetResetToLog
(
&
dataRsp
.
rspOffset
,
walGetLastVer
(
pTq
->
pVnode
->
pWal
));
tqDebug
(
"tmq poll: consumer %"
PRId64
", subkey %s, vg %d, offset reset to %"
PRId64
,
consumerId
,
tqDebug
(
"tmq poll: consumer %"
PRId64
", subkey %s, vg %d, offset reset to %"
PRId64
,
consumerId
,
pHandle
->
subKey
,
TD_VID
(
pTq
->
pVnode
),
dataRsp
.
rspOffset
.
version
);
pHandle
->
subKey
,
TD_VID
(
pTq
->
pVnode
),
dataRsp
.
rspOffset
.
version
);
if
(
tqSendDataRsp
(
pTq
,
pMsg
,
pR
eq
,
&
dataRsp
)
<
0
)
{
if
(
tqSendDataRsp
(
pTq
,
pMsg
,
&
r
eq
,
&
dataRsp
)
<
0
)
{
code
=
-
1
;
code
=
-
1
;
}
}
tDeleteSMqDataRsp
(
&
dataRsp
);
tDeleteSMqDataRsp
(
&
dataRsp
);
return
code
;
return
code
;
}
else
{
}
else
{
STaosxRsp
taosxRsp
=
{
0
};
STaosxRsp
taosxRsp
=
{
0
};
tqInitTaosxRsp
(
&
taosxRsp
,
pR
eq
);
tqInitTaosxRsp
(
&
taosxRsp
,
&
r
eq
);
tqOffsetResetToLog
(
&
taosxRsp
.
rspOffset
,
walGetLastVer
(
pTq
->
pVnode
->
pWal
));
tqOffsetResetToLog
(
&
taosxRsp
.
rspOffset
,
walGetLastVer
(
pTq
->
pVnode
->
pWal
));
if
(
tqSendTaosxRsp
(
pTq
,
pMsg
,
pR
eq
,
&
taosxRsp
)
<
0
)
{
if
(
tqSendTaosxRsp
(
pTq
,
pMsg
,
&
r
eq
,
&
taosxRsp
)
<
0
)
{
code
=
-
1
;
code
=
-
1
;
}
}
tDeleteSTaosxRsp
(
&
taosxRsp
);
tDeleteSTaosxRsp
(
&
taosxRsp
);
...
@@ -543,7 +549,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -543,7 +549,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
}
else
if
(
reqOffset
.
type
==
TMQ_OFFSET__RESET_NONE
)
{
}
else
if
(
reqOffset
.
type
==
TMQ_OFFSET__RESET_NONE
)
{
tqError
(
"tmq poll: subkey %s, no offset committed for consumer %"
PRId64
tqError
(
"tmq poll: subkey %s, no offset committed for consumer %"
PRId64
" in vg %d, subkey %s, reset none failed"
,
" in vg %d, subkey %s, reset none failed"
,
pHandle
->
subKey
,
consumerId
,
TD_VID
(
pTq
->
pVnode
),
pReq
->
subKey
);
pHandle
->
subKey
,
consumerId
,
TD_VID
(
pTq
->
pVnode
),
req
.
subKey
);
terrno
=
TSDB_CODE_TQ_NO_COMMITTED_OFFSET
;
terrno
=
TSDB_CODE_TQ_NO_COMMITTED_OFFSET
;
return
-
1
;
return
-
1
;
}
}
...
@@ -552,7 +558,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -552,7 +558,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
if
(
pHandle
->
execHandle
.
subType
==
TOPIC_SUB_TYPE__COLUMN
)
{
if
(
pHandle
->
execHandle
.
subType
==
TOPIC_SUB_TYPE__COLUMN
)
{
SMqDataRsp
dataRsp
=
{
0
};
SMqDataRsp
dataRsp
=
{
0
};
tqInitDataRsp
(
&
dataRsp
,
pR
eq
,
pHandle
->
execHandle
.
subType
);
tqInitDataRsp
(
&
dataRsp
,
&
r
eq
,
pHandle
->
execHandle
.
subType
);
// lock
// lock
taosWLockLatch
(
&
pTq
->
pushLock
);
taosWLockLatch
(
&
pTq
->
pushLock
);
tqScanData
(
pTq
,
pHandle
,
&
dataRsp
,
&
fetchOffsetNew
);
tqScanData
(
pTq
,
pHandle
,
&
dataRsp
,
&
fetchOffsetNew
);
...
@@ -580,7 +586,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -580,7 +586,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
#endif
#endif
taosWUnLockLatch
(
&
pTq
->
pushLock
);
taosWUnLockLatch
(
&
pTq
->
pushLock
);
if
(
tqSendDataRsp
(
pTq
,
pMsg
,
pR
eq
,
&
dataRsp
)
<
0
)
{
if
(
tqSendDataRsp
(
pTq
,
pMsg
,
&
r
eq
,
&
dataRsp
)
<
0
)
{
code
=
-
1
;
code
=
-
1
;
}
}
...
@@ -599,13 +605,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -599,13 +605,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
SMqMetaRsp
metaRsp
=
{
0
};
SMqMetaRsp
metaRsp
=
{
0
};
STaosxRsp
taosxRsp
=
{
0
};
STaosxRsp
taosxRsp
=
{
0
};
tqInitTaosxRsp
(
&
taosxRsp
,
pR
eq
);
tqInitTaosxRsp
(
&
taosxRsp
,
&
r
eq
);
if
(
fetchOffsetNew
.
type
!=
TMQ_OFFSET__LOG
)
{
if
(
fetchOffsetNew
.
type
!=
TMQ_OFFSET__LOG
)
{
tqScanTaosx
(
pTq
,
pHandle
,
&
taosxRsp
,
&
metaRsp
,
&
fetchOffsetNew
);
tqScanTaosx
(
pTq
,
pHandle
,
&
taosxRsp
,
&
metaRsp
,
&
fetchOffsetNew
);
if
(
metaRsp
.
metaRspLen
>
0
)
{
if
(
metaRsp
.
metaRspLen
>
0
)
{
if
(
tqSendMetaPollRsp
(
pTq
,
pMsg
,
pR
eq
,
&
metaRsp
)
<
0
)
{
if
(
tqSendMetaPollRsp
(
pTq
,
pMsg
,
&
r
eq
,
&
metaRsp
)
<
0
)
{
code
=
-
1
;
code
=
-
1
;
}
}
tqDebug
(
"tmq poll: consumer %"
PRId64
", subkey %s, vg %d, send meta offset type:%d,uid:%"
PRId64
tqDebug
(
"tmq poll: consumer %"
PRId64
", subkey %s, vg %d, send meta offset type:%d,uid:%"
PRId64
...
@@ -618,7 +624,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -618,7 +624,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
}
}
if
(
taosxRsp
.
blockNum
>
0
)
{
if
(
taosxRsp
.
blockNum
>
0
)
{
if
(
tqSendTaosxRsp
(
pTq
,
pMsg
,
pR
eq
,
&
taosxRsp
)
<
0
)
{
if
(
tqSendTaosxRsp
(
pTq
,
pMsg
,
&
r
eq
,
&
taosxRsp
)
<
0
)
{
code
=
-
1
;
code
=
-
1
;
}
}
tDeleteSTaosxRsp
(
&
taosxRsp
);
tDeleteSTaosxRsp
(
&
taosxRsp
);
...
@@ -648,13 +654,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -648,13 +654,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
if
(
consumerEpoch
>
reqEpoch
)
{
if
(
consumerEpoch
>
reqEpoch
)
{
tqWarn
(
"tmq poll: consumer %"
PRId64
" (epoch %d), subkey %s, vg %d offset %"
PRId64
tqWarn
(
"tmq poll: consumer %"
PRId64
" (epoch %d), subkey %s, vg %d offset %"
PRId64
", found new consumer epoch %d, discard req epoch %d"
,
", found new consumer epoch %d, discard req epoch %d"
,
consumerId
,
pReq
->
epoch
,
pHandle
->
subKey
,
TD_VID
(
pTq
->
pVnode
),
fetchVer
,
consumerEpoch
,
reqEpoch
);
consumerId
,
req
.
epoch
,
pHandle
->
subKey
,
TD_VID
(
pTq
->
pVnode
),
fetchVer
,
consumerEpoch
,
reqEpoch
);
break
;
break
;
}
}
if
(
tqFetchLog
(
pTq
,
pHandle
,
&
fetchVer
,
&
pCkHead
)
<
0
)
{
if
(
tqFetchLog
(
pTq
,
pHandle
,
&
fetchVer
,
&
pCkHead
)
<
0
)
{
tqOffsetResetToLog
(
&
taosxRsp
.
rspOffset
,
fetchVer
);
tqOffsetResetToLog
(
&
taosxRsp
.
rspOffset
,
fetchVer
);
if
(
tqSendTaosxRsp
(
pTq
,
pMsg
,
pR
eq
,
&
taosxRsp
)
<
0
)
{
if
(
tqSendTaosxRsp
(
pTq
,
pMsg
,
&
r
eq
,
&
taosxRsp
)
<
0
)
{
code
=
-
1
;
code
=
-
1
;
}
}
tDeleteSTaosxRsp
(
&
taosxRsp
);
tDeleteSTaosxRsp
(
&
taosxRsp
);
...
@@ -665,7 +671,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -665,7 +671,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
SWalCont
*
pHead
=
&
pCkHead
->
head
;
SWalCont
*
pHead
=
&
pCkHead
->
head
;
tqDebug
(
"tmq poll: consumer:%"
PRId64
", (epoch %d) iter log, vgId:%d offset %"
PRId64
" msgType %d"
,
consumerId
,
tqDebug
(
"tmq poll: consumer:%"
PRId64
", (epoch %d) iter log, vgId:%d offset %"
PRId64
" msgType %d"
,
consumerId
,
pReq
->
epoch
,
TD_VID
(
pTq
->
pVnode
),
fetchVer
,
pHead
->
msgType
);
req
.
epoch
,
TD_VID
(
pTq
->
pVnode
),
fetchVer
,
pHead
->
msgType
);
if
(
pHead
->
msgType
==
TDMT_VND_SUBMIT
)
{
if
(
pHead
->
msgType
==
TDMT_VND_SUBMIT
)
{
SSubmitReq
*
pCont
=
(
SSubmitReq
*
)
&
pHead
->
body
;
SSubmitReq
*
pCont
=
(
SSubmitReq
*
)
&
pHead
->
body
;
...
@@ -674,7 +680,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -674,7 +680,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
}
}
if
(
taosxRsp
.
blockNum
>
0
/* threshold */
)
{
if
(
taosxRsp
.
blockNum
>
0
/* threshold */
)
{
tqOffsetResetToLog
(
&
taosxRsp
.
rspOffset
,
fetchVer
);
tqOffsetResetToLog
(
&
taosxRsp
.
rspOffset
,
fetchVer
);
if
(
tqSendTaosxRsp
(
pTq
,
pMsg
,
pR
eq
,
&
taosxRsp
)
<
0
)
{
if
(
tqSendTaosxRsp
(
pTq
,
pMsg
,
&
r
eq
,
&
taosxRsp
)
<
0
)
{
code
=
-
1
;
code
=
-
1
;
}
}
tDeleteSTaosxRsp
(
&
taosxRsp
);
tDeleteSTaosxRsp
(
&
taosxRsp
);
...
@@ -692,7 +698,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
...
@@ -692,7 +698,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
metaRsp
.
resMsgType
=
pHead
->
msgType
;
metaRsp
.
resMsgType
=
pHead
->
msgType
;
metaRsp
.
metaRspLen
=
pHead
->
bodyLen
;
metaRsp
.
metaRspLen
=
pHead
->
bodyLen
;
metaRsp
.
metaRsp
=
pHead
->
body
;
metaRsp
.
metaRsp
=
pHead
->
body
;
if
(
tqSendMetaPollRsp
(
pTq
,
pMsg
,
pR
eq
,
&
metaRsp
)
<
0
)
{
if
(
tqSendMetaPollRsp
(
pTq
,
pMsg
,
&
r
eq
,
&
metaRsp
)
<
0
)
{
code
=
-
1
;
code
=
-
1
;
taosMemoryFreeClear
(
pCkHead
);
taosMemoryFreeClear
(
pCkHead
);
tDeleteSTaosxRsp
(
&
taosxRsp
);
tDeleteSTaosxRsp
(
&
taosxRsp
);
...
...
source/libs/executor/src/exchangeoperator.c
浏览文件 @
997a00ce
...
@@ -416,27 +416,42 @@ int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTas
...
@@ -416,27 +416,42 @@ int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTas
loadRemoteDataCallback
(
pWrapper
,
&
pBuf
,
code
);
loadRemoteDataCallback
(
pWrapper
,
&
pBuf
,
code
);
taosMemoryFree
(
pWrapper
);
taosMemoryFree
(
pWrapper
);
}
else
{
}
else
{
SResFetchReq
*
pMsg
=
taosMemoryCalloc
(
1
,
sizeof
(
SResFetchReq
));
SResFetchReq
req
=
{
0
};
if
(
NULL
==
pMsg
)
{
req
.
header
.
vgId
=
pSource
->
addr
.
nodeId
;
req
.
sId
=
pSource
->
schedId
;
req
.
taskId
=
pSource
->
taskId
;
req
.
queryId
=
pTaskInfo
->
id
.
queryId
;
req
.
execId
=
pSource
->
execId
;
int32_t
msgSize
=
tSerializeSResFetchReq
(
NULL
,
0
,
&
req
);
if
(
msgSize
<
0
)
{
pTaskInfo
->
code
=
TSDB_CODE_QRY_OUT_OF_MEMORY
;
pTaskInfo
->
code
=
TSDB_CODE_QRY_OUT_OF_MEMORY
;
taosMemoryFree
(
pWrapper
);
taosMemoryFree
(
pWrapper
);
return
pTaskInfo
->
code
;
return
pTaskInfo
->
code
;
}
}
void
*
msg
=
taosMemoryCalloc
(
1
,
msgSize
);
if
(
NULL
==
msg
)
{
pTaskInfo
->
code
=
TSDB_CODE_QRY_OUT_OF_MEMORY
;
taosMemoryFree
(
pWrapper
);
return
pTaskInfo
->
code
;
}
if
(
tSerializeSResFetchReq
(
msg
,
msgSize
,
&
req
)
<
0
)
{
pTaskInfo
->
code
=
TSDB_CODE_QRY_OUT_OF_MEMORY
;
taosMemoryFree
(
pWrapper
);
taosMemoryFree
(
msg
);
return
pTaskInfo
->
code
;
}
qDebug
(
"%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%"
PRIx64
", execId:%d, %d/%"
PRIzu
,
qDebug
(
"%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%"
PRIx64
", execId:%d, %d/%"
PRIzu
,
GET_TASKID
(
pTaskInfo
),
pSource
->
addr
.
nodeId
,
pSource
->
addr
.
epSet
.
eps
[
0
].
fqdn
,
pSource
->
taskId
,
GET_TASKID
(
pTaskInfo
),
pSource
->
addr
.
nodeId
,
pSource
->
addr
.
epSet
.
eps
[
0
].
fqdn
,
pSource
->
taskId
,
pSource
->
execId
,
sourceIndex
,
totalSources
);
pSource
->
execId
,
sourceIndex
,
totalSources
);
pMsg
->
header
.
vgId
=
htonl
(
pSource
->
addr
.
nodeId
);
pMsg
->
sId
=
htobe64
(
pSource
->
schedId
);
pMsg
->
taskId
=
htobe64
(
pSource
->
taskId
);
pMsg
->
queryId
=
htobe64
(
pTaskInfo
->
id
.
queryId
);
pMsg
->
execId
=
htonl
(
pSource
->
execId
);
// send the fetch remote task result reques
// send the fetch remote task result reques
SMsgSendInfo
*
pMsgSendInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SMsgSendInfo
));
SMsgSendInfo
*
pMsgSendInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SMsgSendInfo
));
if
(
NULL
==
pMsgSendInfo
)
{
if
(
NULL
==
pMsgSendInfo
)
{
taosMemoryFreeClear
(
pM
sg
);
taosMemoryFreeClear
(
m
sg
);
taosMemoryFree
(
pWrapper
);
taosMemoryFree
(
pWrapper
);
qError
(
"%s prepare message %d failed"
,
GET_TASKID
(
pTaskInfo
),
(
int32_t
)
sizeof
(
SMsgSendInfo
));
qError
(
"%s prepare message %d failed"
,
GET_TASKID
(
pTaskInfo
),
(
int32_t
)
sizeof
(
SMsgSendInfo
));
pTaskInfo
->
code
=
TSDB_CODE_QRY_OUT_OF_MEMORY
;
pTaskInfo
->
code
=
TSDB_CODE_QRY_OUT_OF_MEMORY
;
...
@@ -445,8 +460,8 @@ int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTas
...
@@ -445,8 +460,8 @@ int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTas
pMsgSendInfo
->
param
=
pWrapper
;
pMsgSendInfo
->
param
=
pWrapper
;
pMsgSendInfo
->
paramFreeFp
=
taosMemoryFree
;
pMsgSendInfo
->
paramFreeFp
=
taosMemoryFree
;
pMsgSendInfo
->
msgInfo
.
pData
=
pM
sg
;
pMsgSendInfo
->
msgInfo
.
pData
=
m
sg
;
pMsgSendInfo
->
msgInfo
.
len
=
sizeof
(
SResFetchReq
)
;
pMsgSendInfo
->
msgInfo
.
len
=
msgSize
;
pMsgSendInfo
->
msgType
=
pSource
->
fetchMsgType
;
pMsgSendInfo
->
msgType
=
pSource
->
fetchMsgType
;
pMsgSendInfo
->
fp
=
loadRemoteDataCallback
;
pMsgSendInfo
->
fp
=
loadRemoteDataCallback
;
...
...
source/libs/qworker/src/qwMsg.c
浏览文件 @
997a00ce
...
@@ -499,27 +499,22 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int
...
@@ -499,27 +499,22 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int
return
TSDB_CODE_QRY_INVALID_INPUT
;
return
TSDB_CODE_QRY_INVALID_INPUT
;
}
}
SResFetchReq
*
msg
=
pMsg
->
pCont
;
SResFetchReq
req
=
{
0
}
;
SQWorker
*
mgmt
=
(
SQWorker
*
)
qWorkerMgmt
;
SQWorker
*
mgmt
=
(
SQWorker
*
)
qWorkerMgmt
;
qwUpdateTimeInQueue
(
mgmt
,
ts
,
FETCH_QUEUE
);
qwUpdateTimeInQueue
(
mgmt
,
ts
,
FETCH_QUEUE
);
QW_STAT_INC
(
mgmt
->
stat
.
msgStat
.
fetchProcessed
,
1
);
QW_STAT_INC
(
mgmt
->
stat
.
msgStat
.
fetchProcessed
,
1
);
if
(
NULL
==
msg
||
pMsg
->
contLen
<
sizeof
(
*
msg
)
)
{
if
(
tDeserializeSResFetchReq
(
pMsg
->
pCont
,
pMsg
->
contLen
,
&
req
)
<
0
)
{
QW_ELOG
(
"
invalid fetch msg, msg:%p, msgLen:%d"
,
msg
,
pMsg
->
contLen
);
QW_ELOG
(
"
tDeserializeSResFetchReq %d failed"
,
pMsg
->
contLen
);
QW_ERR_RET
(
TSDB_CODE_QRY_INVALID_INPUT
);
QW_ERR_RET
(
TSDB_CODE_QRY_INVALID_INPUT
);
}
}
msg
->
sId
=
be64toh
(
msg
->
sId
);
uint64_t
sId
=
req
.
sId
;
msg
->
queryId
=
be64toh
(
msg
->
queryId
);
uint64_t
qId
=
req
.
queryId
;
msg
->
taskId
=
be64toh
(
msg
->
taskId
);
uint64_t
tId
=
req
.
taskId
;
msg
->
execId
=
ntohl
(
msg
->
execId
);
uint64_t
sId
=
msg
->
sId
;
uint64_t
qId
=
msg
->
queryId
;
uint64_t
tId
=
msg
->
taskId
;
int64_t
rId
=
0
;
int64_t
rId
=
0
;
int32_t
eId
=
msg
->
execId
;
int32_t
eId
=
req
.
execId
;
SQWMsg
qwMsg
=
{.
node
=
node
,
.
msg
=
NULL
,
.
msgLen
=
0
,
.
connInfo
=
pMsg
->
info
,
.
msgType
=
pMsg
->
msgType
};
SQWMsg
qwMsg
=
{.
node
=
node
,
.
msg
=
NULL
,
.
msgLen
=
0
,
.
connInfo
=
pMsg
->
info
,
.
msgType
=
pMsg
->
msgType
};
...
...
source/libs/scheduler/src/schRemote.c
浏览文件 @
997a00ce
...
@@ -1083,22 +1083,29 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
...
@@ -1083,22 +1083,29 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
}
}
case
TDMT_SCH_FETCH
:
case
TDMT_SCH_FETCH
:
case
TDMT_SCH_MERGE_FETCH
:
{
case
TDMT_SCH_MERGE_FETCH
:
{
msgSize
=
sizeof
(
SResFetchReq
);
SResFetchReq
req
=
{
0
};
req
.
header
.
vgId
=
addr
->
nodeId
;
req
.
sId
=
schMgmt
.
sId
;
req
.
queryId
=
pJob
->
queryId
;
req
.
taskId
=
pTask
->
taskId
;
req
.
execId
=
pTask
->
execId
;
msgSize
=
tSerializeSResFetchReq
(
NULL
,
0
,
&
req
);
if
(
msgSize
<
0
)
{
SCH_TASK_ELOG
(
"tSerializeSResFetchReq get size, msgSize:%d"
,
msgSize
);
SCH_ERR_RET
(
TSDB_CODE_QRY_OUT_OF_MEMORY
);
}
msg
=
taosMemoryCalloc
(
1
,
msgSize
);
msg
=
taosMemoryCalloc
(
1
,
msgSize
);
if
(
NULL
==
msg
)
{
if
(
NULL
==
msg
)
{
SCH_TASK_ELOG
(
"calloc %d failed"
,
msgSize
);
SCH_TASK_ELOG
(
"calloc %d failed"
,
msgSize
);
SCH_ERR_RET
(
TSDB_CODE_QRY_OUT_OF_MEMORY
);
SCH_ERR_RET
(
TSDB_CODE_QRY_OUT_OF_MEMORY
);
}
}
SResFetchReq
*
pMsg
=
msg
;
if
(
tSerializeSResFetchReq
(
msg
,
msgSize
,
&
req
)
<
0
)
{
SCH_TASK_ELOG
(
"tSerializeSResFetchReq %d failed"
,
msgSize
);
pMsg
->
header
.
vgId
=
htonl
(
addr
->
nodeId
);
SCH_ERR_RET
(
TSDB_CODE_QRY_OUT_OF_MEMORY
);
}
pMsg
->
sId
=
htobe64
(
schMgmt
.
sId
);
pMsg
->
queryId
=
htobe64
(
pJob
->
queryId
);
pMsg
->
taskId
=
htobe64
(
pTask
->
taskId
);
pMsg
->
execId
=
htonl
(
pTask
->
execId
);
break
;
break
;
}
}
case
TDMT_SCH_DROP_TASK
:
{
case
TDMT_SCH_DROP_TASK
:
{
...
...
tests/script/runAllSimCases.sh
浏览文件 @
997a00ce
...
@@ -11,6 +11,9 @@ set -e
...
@@ -11,6 +11,9 @@ set -e
VALGRIND
=
0
VALGRIND
=
0
LOG_BK_DIR
=
/data/valgrind_log_backup
# 192.168.0.203
LOG_BK_DIR
=
/data/valgrind_log_backup
# 192.168.0.203
SIM_FILES
=
./jenkins/basic.txt
SIM_FILES
=
./jenkins/basic.txt
cases_task_file
=
../parallel_test/cases.task
cat
$cases_task_file
|
grep
"./test.sh "
|
awk
-F
,
'{print $5}'
>
./jenkins/basic.txt
while
getopts
"v:r:f:"
arg
while
getopts
"v:r:f:"
arg
do
do
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录