Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
323ce214
T
TDengine
项目概览
taosdata
/
TDengine
接近 2 年 前同步成功
通知
1192
Star
22018
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看板
提交
323ce214
编写于
5月 07, 2022
作者:
H
Hongze Cheng
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refact encode APIs
上级
4a9ae285
变更
27
展开全部
隐藏空白更改
内联
并排
Showing
27 changed file
with
943 addition
and
940 deletion
+943
-940
include/common/tmsg.h
include/common/tmsg.h
+36
-36
include/libs/stream/tstream.h
include/libs/stream/tstream.h
+2
-2
include/util/tencode.h
include/util/tencode.h
+200
-206
source/client/src/tmq.c
source/client/src/tmq.c
+6
-6
source/common/src/tmsg.c
source/common/src/tmsg.c
+466
-466
source/dnode/mnode/impl/inc/mndDef.h
source/dnode/mnode/impl/inc/mndDef.h
+2
-2
source/dnode/mnode/impl/src/mndDef.c
source/dnode/mnode/impl/src/mndDef.c
+2
-2
source/dnode/mnode/impl/src/mndOffset.c
source/dnode/mnode/impl/src/mndOffset.c
+2
-2
source/dnode/mnode/impl/src/mndScheduler.c
source/dnode/mnode/impl/src/mndScheduler.c
+5
-5
source/dnode/mnode/impl/src/mndStb.c
source/dnode/mnode/impl/src/mndStb.c
+8
-8
source/dnode/mnode/impl/src/mndStream.c
source/dnode/mnode/impl/src/mndStream.c
+9
-9
source/dnode/snode/src/snode.c
source/dnode/snode/src/snode.c
+3
-3
source/dnode/vnode/inc/vnode.h
source/dnode/vnode/inc/vnode.h
+1
-1
source/dnode/vnode/src/inc/meta.h
source/dnode/vnode/src/inc/meta.h
+2
-2
source/dnode/vnode/src/meta/metaEntry.c
source/dnode/vnode/src/meta/metaEntry.c
+2
-2
source/dnode/vnode/src/meta/metaQuery.c
source/dnode/vnode/src/meta/metaQuery.c
+5
-5
source/dnode/vnode/src/meta/metaTable.c
source/dnode/vnode/src/meta/metaTable.c
+9
-10
source/dnode/vnode/src/tq/tq.c
source/dnode/vnode/src/tq/tq.c
+3
-3
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
+1
-1
source/dnode/vnode/src/vnd/vnodeSvr.c
source/dnode/vnode/src/vnd/vnodeSvr.c
+29
-26
source/libs/monitor/src/monMsg.c
source/libs/monitor/src/monMsg.c
+58
-58
source/libs/parser/src/parInsertData.c
source/libs/parser/src/parInsertData.c
+3
-3
source/libs/parser/src/parTranslater.c
source/libs/parser/src/parTranslater.c
+8
-8
source/libs/scheduler/src/scheduler.c
source/libs/scheduler/src/scheduler.c
+8
-8
source/libs/stream/src/tstream.c
source/libs/stream/src/tstream.c
+2
-2
source/libs/sync/src/syncMessage.c
source/libs/sync/src/syncMessage.c
+12
-12
source/util/src/tencode.c
source/util/src/tencode.c
+59
-52
未找到文件。
include/common/tmsg.h
浏览文件 @
323ce214
...
...
@@ -338,8 +338,8 @@ typedef struct SEpSet {
SEp
eps
[
TSDB_MAX_REPLICA
];
}
SEpSet
;
int32_t
tEncodeSEpSet
(
S
C
oder
*
pEncoder
,
const
SEpSet
*
pEp
);
int32_t
tDecodeSEpSet
(
S
C
oder
*
pDecoder
,
SEpSet
*
pEp
);
int32_t
tEncodeSEpSet
(
S
Enc
oder
*
pEncoder
,
const
SEpSet
*
pEp
);
int32_t
tDecodeSEpSet
(
S
Dec
oder
*
pDecoder
,
SEpSet
*
pEp
);
int32_t
taosEncodeSEpSet
(
void
**
buf
,
const
SEpSet
*
pEp
);
void
*
taosDecodeSEpSet
(
const
void
*
buf
,
SEpSet
*
pEp
);
...
...
@@ -611,8 +611,8 @@ typedef struct {
int32_t
tSerializeSUseDbRsp
(
void
*
buf
,
int32_t
bufLen
,
const
SUseDbRsp
*
pRsp
);
int32_t
tDeserializeSUseDbRsp
(
void
*
buf
,
int32_t
bufLen
,
SUseDbRsp
*
pRsp
);
int32_t
tSerializeSUseDbRspImp
(
S
C
oder
*
pEncoder
,
const
SUseDbRsp
*
pRsp
);
int32_t
tDeserializeSUseDbRspImp
(
S
C
oder
*
pDecoder
,
SUseDbRsp
*
pRsp
);
int32_t
tSerializeSUseDbRspImp
(
S
Enc
oder
*
pEncoder
,
const
SUseDbRsp
*
pRsp
);
int32_t
tDeserializeSUseDbRspImp
(
S
Dec
oder
*
pDecoder
,
SUseDbRsp
*
pRsp
);
void
tFreeSUsedbRsp
(
SUseDbRsp
*
pRsp
);
typedef
struct
{
...
...
@@ -1527,8 +1527,8 @@ typedef struct {
char
*
qmsg2
;
// pAst2:qmsg2:SRetention2 => trigger aggr task2
}
SRSmaParam
;
int32_t
tEncodeSRSmaParam
(
S
C
oder
*
pCoder
,
const
SRSmaParam
*
pRSmaParam
);
int32_t
tDecodeSRSmaParam
(
S
C
oder
*
pCoder
,
SRSmaParam
*
pRSmaParam
);
int32_t
tEncodeSRSmaParam
(
S
Enc
oder
*
pCoder
,
const
SRSmaParam
*
pRSmaParam
);
int32_t
tDecodeSRSmaParam
(
S
Dec
oder
*
pCoder
,
SRSmaParam
*
pRSmaParam
);
// TDMT_VND_CREATE_STB ==============
typedef
struct
SVCreateStbReq
{
...
...
@@ -1540,8 +1540,8 @@ typedef struct SVCreateStbReq {
SRSmaParam
pRSmaParam
;
}
SVCreateStbReq
;
int
tEncodeSVCreateStbReq
(
S
C
oder
*
pCoder
,
const
SVCreateStbReq
*
pReq
);
int
tDecodeSVCreateStbReq
(
S
C
oder
*
pCoder
,
SVCreateStbReq
*
pReq
);
int
tEncodeSVCreateStbReq
(
S
Enc
oder
*
pCoder
,
const
SVCreateStbReq
*
pReq
);
int
tDecodeSVCreateStbReq
(
S
Dec
oder
*
pCoder
,
SVCreateStbReq
*
pReq
);
// TDMT_VND_DROP_STB ==============
typedef
struct
SVDropStbReq
{
...
...
@@ -1549,8 +1549,8 @@ typedef struct SVDropStbReq {
tb_uid_t
suid
;
}
SVDropStbReq
;
int32_t
tEncodeSVDropStbReq
(
S
C
oder
*
pCoder
,
const
SVDropStbReq
*
pReq
);
int32_t
tDecodeSVDropStbReq
(
S
C
oder
*
pCoder
,
SVDropStbReq
*
pReq
);
int32_t
tEncodeSVDropStbReq
(
S
Enc
oder
*
pCoder
,
const
SVDropStbReq
*
pReq
);
int32_t
tDecodeSVDropStbReq
(
S
Dec
oder
*
pCoder
,
SVDropStbReq
*
pReq
);
#define TD_CREATE_IF_NOT_EXISTS 0x1
typedef
struct
SVCreateTbReq
{
...
...
@@ -1571,8 +1571,8 @@ typedef struct SVCreateTbReq {
};
}
SVCreateTbReq
;
int
tEncodeSVCreateTbReq
(
S
C
oder
*
pCoder
,
const
SVCreateTbReq
*
pReq
);
int
tDecodeSVCreateTbReq
(
S
C
oder
*
pCoder
,
SVCreateTbReq
*
pReq
);
int
tEncodeSVCreateTbReq
(
S
Enc
oder
*
pCoder
,
const
SVCreateTbReq
*
pReq
);
int
tDecodeSVCreateTbReq
(
S
Dec
oder
*
pCoder
,
SVCreateTbReq
*
pReq
);
typedef
struct
{
int32_t
nReqs
;
...
...
@@ -1582,15 +1582,15 @@ typedef struct {
};
}
SVCreateTbBatchReq
;
int
tEncodeSVCreateTbBatchReq
(
S
C
oder
*
pCoder
,
const
SVCreateTbBatchReq
*
pReq
);
int
tDecodeSVCreateTbBatchReq
(
S
C
oder
*
pCoder
,
SVCreateTbBatchReq
*
pReq
);
int
tEncodeSVCreateTbBatchReq
(
S
Enc
oder
*
pCoder
,
const
SVCreateTbBatchReq
*
pReq
);
int
tDecodeSVCreateTbBatchReq
(
S
Dec
oder
*
pCoder
,
SVCreateTbBatchReq
*
pReq
);
typedef
struct
{
int32_t
code
;
}
SVCreateTbRsp
,
SVUpdateTbRsp
;
int
tEncodeSVCreateTbRsp
(
S
C
oder
*
pCoder
,
const
SVCreateTbRsp
*
pRsp
);
int
tDecodeSVCreateTbRsp
(
S
C
oder
*
pCoder
,
SVCreateTbRsp
*
pRsp
);
int
tEncodeSVCreateTbRsp
(
S
Enc
oder
*
pCoder
,
const
SVCreateTbRsp
*
pRsp
);
int
tDecodeSVCreateTbRsp
(
S
Dec
oder
*
pCoder
,
SVCreateTbRsp
*
pRsp
);
int32_t
tSerializeSVCreateTbReq
(
void
**
buf
,
SVCreateTbReq
*
pReq
);
void
*
tDeserializeSVCreateTbReq
(
void
*
buf
,
SVCreateTbReq
*
pReq
);
...
...
@@ -1603,8 +1603,8 @@ typedef struct {
};
}
SVCreateTbBatchRsp
;
int
tEncodeSVCreateTbBatchRsp
(
S
C
oder
*
pCoder
,
const
SVCreateTbBatchRsp
*
pRsp
);
int
tDecodeSVCreateTbBatchRsp
(
S
C
oder
*
pCoder
,
SVCreateTbBatchRsp
*
pRsp
);
int
tEncodeSVCreateTbBatchRsp
(
S
Enc
oder
*
pCoder
,
const
SVCreateTbBatchRsp
*
pRsp
);
int
tDecodeSVCreateTbBatchRsp
(
S
Dec
oder
*
pCoder
,
SVCreateTbBatchRsp
*
pRsp
);
int32_t
tSerializeSVCreateTbBatchRsp
(
void
*
buf
,
int32_t
bufLen
,
SVCreateTbBatchRsp
*
pRsp
);
int32_t
tDeserializeSVCreateTbBatchRsp
(
void
*
buf
,
int32_t
bufLen
,
SVCreateTbBatchRsp
*
pRsp
);
...
...
@@ -1627,8 +1627,8 @@ typedef struct {
};
}
SVDropTbBatchReq
;
int32_t
tEncodeSVDropTbBatchReq
(
S
C
oder
*
pCoder
,
const
SVDropTbBatchReq
*
pReq
);
int32_t
tDecodeSVDropTbBatchReq
(
S
C
oder
*
pCoder
,
SVDropTbBatchReq
*
pReq
);
int32_t
tEncodeSVDropTbBatchReq
(
S
Enc
oder
*
pCoder
,
const
SVDropTbBatchReq
*
pReq
);
int32_t
tDecodeSVDropTbBatchReq
(
S
Dec
oder
*
pCoder
,
SVDropTbBatchReq
*
pReq
);
typedef
struct
{
int32_t
nRsps
;
...
...
@@ -1638,8 +1638,8 @@ typedef struct {
};
}
SVDropTbBatchRsp
;
int32_t
tEncodeSVDropTbBatchRsp
(
S
C
oder
*
pCoder
,
const
SVDropTbBatchRsp
*
pRsp
);
int32_t
tDecodeSVDropTbBatchRsp
(
S
C
oder
*
pCoder
,
SVDropTbBatchRsp
*
pRsp
);
int32_t
tEncodeSVDropTbBatchRsp
(
S
Enc
oder
*
pCoder
,
const
SVDropTbBatchRsp
*
pRsp
);
int32_t
tDecodeSVDropTbBatchRsp
(
S
Dec
oder
*
pCoder
,
SVDropTbBatchRsp
*
pRsp
);
typedef
struct
{
SMsgHead
head
;
...
...
@@ -1821,14 +1821,14 @@ static FORCE_INLINE void tFreeClientHbBatchRsp(void* pRsp) {
int32_t
tSerializeSClientHbBatchRsp
(
void
*
buf
,
int32_t
bufLen
,
const
SClientHbBatchRsp
*
pBatchRsp
);
int32_t
tDeserializeSClientHbBatchRsp
(
void
*
buf
,
int32_t
bufLen
,
SClientHbBatchRsp
*
pBatchRsp
);
static
FORCE_INLINE
int32_t
tEncodeSKv
(
S
C
oder
*
pEncoder
,
const
SKv
*
pKv
)
{
static
FORCE_INLINE
int32_t
tEncodeSKv
(
S
Enc
oder
*
pEncoder
,
const
SKv
*
pKv
)
{
if
(
tEncodeI32
(
pEncoder
,
pKv
->
key
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
pEncoder
,
pKv
->
valueLen
)
<
0
)
return
-
1
;
if
(
tEncodeBinary
(
pEncoder
,
(
const
char
*
)
pKv
->
value
,
pKv
->
valueLen
)
<
0
)
return
-
1
;
return
0
;
}
static
FORCE_INLINE
int32_t
tDecodeSKv
(
S
C
oder
*
pDecoder
,
SKv
*
pKv
)
{
static
FORCE_INLINE
int32_t
tDecodeSKv
(
S
Dec
oder
*
pDecoder
,
SKv
*
pKv
)
{
if
(
tDecodeI32
(
pDecoder
,
&
pKv
->
key
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
pDecoder
,
&
pKv
->
valueLen
)
<
0
)
return
-
1
;
pKv
->
value
=
taosMemoryMalloc
(
pKv
->
valueLen
+
1
);
...
...
@@ -1837,13 +1837,13 @@ static FORCE_INLINE int32_t tDecodeSKv(SCoder* pDecoder, SKv* pKv) {
return
0
;
}
static
FORCE_INLINE
int32_t
tEncodeSClientHbKey
(
S
C
oder
*
pEncoder
,
const
SClientHbKey
*
pKey
)
{
static
FORCE_INLINE
int32_t
tEncodeSClientHbKey
(
S
Enc
oder
*
pEncoder
,
const
SClientHbKey
*
pKey
)
{
if
(
tEncodeI64
(
pEncoder
,
pKey
->
tscRid
)
<
0
)
return
-
1
;
if
(
tEncodeI8
(
pEncoder
,
pKey
->
connType
)
<
0
)
return
-
1
;
return
0
;
}
static
FORCE_INLINE
int32_t
tDecodeSClientHbKey
(
S
C
oder
*
pDecoder
,
SClientHbKey
*
pKey
)
{
static
FORCE_INLINE
int32_t
tDecodeSClientHbKey
(
S
Dec
oder
*
pDecoder
,
SClientHbKey
*
pKey
)
{
if
(
tDecodeI64
(
pDecoder
,
&
pKey
->
tscRid
)
<
0
)
return
-
1
;
if
(
tDecodeI8
(
pDecoder
,
&
pKey
->
connType
)
<
0
)
return
-
1
;
return
0
;
...
...
@@ -2046,10 +2046,10 @@ typedef struct {
int32_t
reserved
;
}
SMqCMCommitOffsetRsp
;
int32_t
tEncodeSMqOffset
(
S
C
oder
*
encoder
,
const
SMqOffset
*
pOffset
);
int32_t
tDecodeSMqOffset
(
S
C
oder
*
decoder
,
SMqOffset
*
pOffset
);
int32_t
tEncodeSMqCMCommitOffsetReq
(
S
C
oder
*
encoder
,
const
SMqCMCommitOffsetReq
*
pReq
);
int32_t
tDecodeSMqCMCommitOffsetReq
(
S
C
oder
*
decoder
,
SMqCMCommitOffsetReq
*
pReq
);
int32_t
tEncodeSMqOffset
(
S
Enc
oder
*
encoder
,
const
SMqOffset
*
pOffset
);
int32_t
tDecodeSMqOffset
(
S
Dec
oder
*
decoder
,
SMqOffset
*
pOffset
);
int32_t
tEncodeSMqCMCommitOffsetReq
(
S
Enc
oder
*
encoder
,
const
SMqCMCommitOffsetReq
*
pReq
);
int32_t
tDecodeSMqCMCommitOffsetReq
(
S
Dec
oder
*
decoder
,
SMqCMCommitOffsetReq
*
pReq
);
static
FORCE_INLINE
SSchemaWrapper
*
tCloneSSchemaWrapper
(
const
SSchemaWrapper
*
pSchemaWrapper
)
{
SSchemaWrapper
*
pSW
=
(
SSchemaWrapper
*
)
taosMemoryMalloc
(
sizeof
(
SSchemaWrapper
));
...
...
@@ -2089,7 +2089,7 @@ static FORCE_INLINE void* taosDecodeSSchema(const void* buf, SSchema* pSchema) {
return
(
void
*
)
buf
;
}
static
FORCE_INLINE
int32_t
tEncodeSSchema
(
S
C
oder
*
pEncoder
,
const
SSchema
*
pSchema
)
{
static
FORCE_INLINE
int32_t
tEncodeSSchema
(
S
Enc
oder
*
pEncoder
,
const
SSchema
*
pSchema
)
{
if
(
tEncodeI8
(
pEncoder
,
pSchema
->
type
)
<
0
)
return
-
1
;
if
(
tEncodeI8
(
pEncoder
,
pSchema
->
flags
)
<
0
)
return
-
1
;
if
(
tEncodeI32v
(
pEncoder
,
pSchema
->
bytes
)
<
0
)
return
-
1
;
...
...
@@ -2098,7 +2098,7 @@ static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSch
return
0
;
}
static
FORCE_INLINE
int32_t
tDecodeSSchema
(
S
C
oder
*
pDecoder
,
SSchema
*
pSchema
)
{
static
FORCE_INLINE
int32_t
tDecodeSSchema
(
S
Dec
oder
*
pDecoder
,
SSchema
*
pSchema
)
{
if
(
tDecodeI8
(
pDecoder
,
&
pSchema
->
type
)
<
0
)
return
-
1
;
if
(
tDecodeI8
(
pDecoder
,
&
pSchema
->
flags
)
<
0
)
return
-
1
;
if
(
tDecodeI32v
(
pDecoder
,
&
pSchema
->
bytes
)
<
0
)
return
-
1
;
...
...
@@ -2131,7 +2131,7 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapp
return
(
void
*
)
buf
;
}
static
FORCE_INLINE
int32_t
tEncodeSSchemaWrapper
(
S
C
oder
*
pEncoder
,
const
SSchemaWrapper
*
pSW
)
{
static
FORCE_INLINE
int32_t
tEncodeSSchemaWrapper
(
S
Enc
oder
*
pEncoder
,
const
SSchemaWrapper
*
pSW
)
{
if
(
tEncodeI32v
(
pEncoder
,
pSW
->
nCols
)
<
0
)
return
-
1
;
if
(
tEncodeI32v
(
pEncoder
,
pSW
->
sver
)
<
0
)
return
-
1
;
for
(
int32_t
i
=
0
;
i
<
pSW
->
nCols
;
i
++
)
{
...
...
@@ -2141,7 +2141,7 @@ static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchem
return
0
;
}
static
FORCE_INLINE
int32_t
tDecodeSSchemaWrapper
(
S
C
oder
*
pDecoder
,
SSchemaWrapper
*
pSW
)
{
static
FORCE_INLINE
int32_t
tDecodeSSchemaWrapper
(
S
Dec
oder
*
pDecoder
,
SSchemaWrapper
*
pSW
)
{
if
(
tDecodeI32v
(
pDecoder
,
&
pSW
->
nCols
)
<
0
)
return
-
1
;
if
(
tDecodeI32v
(
pDecoder
,
&
pSW
->
sver
)
<
0
)
return
-
1
;
...
...
@@ -2610,8 +2610,8 @@ typedef struct {
};
}
SVSubmitReq
;
int32_t
tEncodeSVSubmitReq
(
S
C
oder
*
pCoder
,
const
SVSubmitReq
*
pReq
);
int32_t
tDecodeSVSubmitReq
(
S
C
oder
*
pCoder
,
SVSubmitReq
*
pReq
);
int32_t
tEncodeSVSubmitReq
(
S
Enc
oder
*
pCoder
,
const
SVSubmitReq
*
pReq
);
int32_t
tDecodeSVSubmitReq
(
S
Dec
oder
*
pCoder
,
SVSubmitReq
*
pReq
);
#pragma pack(pop)
...
...
include/libs/stream/tstream.h
浏览文件 @
323ce214
...
...
@@ -154,8 +154,8 @@ typedef struct {
}
SStreamTask
;
SStreamTask
*
tNewSStreamTask
(
int64_t
streamId
);
int32_t
tEncodeSStreamTask
(
S
C
oder
*
pEncoder
,
const
SStreamTask
*
pTask
);
int32_t
tDecodeSStreamTask
(
S
C
oder
*
pDecoder
,
SStreamTask
*
pTask
);
int32_t
tEncodeSStreamTask
(
S
Enc
oder
*
pEncoder
,
const
SStreamTask
*
pTask
);
int32_t
tDecodeSStreamTask
(
S
Dec
oder
*
pDecoder
,
SStreamTask
*
pTask
);
void
tFreeSStreamTask
(
SStreamTask
*
pTask
);
typedef
struct
{
...
...
include/util/tencode.h
浏览文件 @
323ce214
此差异已折叠。
点击以展开。
source/client/src/tmq.c
浏览文件 @
323ce214
...
...
@@ -547,21 +547,21 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
req
.
offsets
=
(
SMqOffset
*
)
offsets
->
container
.
pData
;
}
S
C
oder
encoder
;
S
Enc
oder
encoder
;
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
NULL
,
0
,
TD_ENCODER
);
t
EncoderInit
(
&
encoder
,
NULL
,
0
);
tEncodeSMqCMCommitOffsetReq
(
&
encoder
,
&
req
);
int32_t
tlen
=
encoder
.
pos
;
void
*
buf
=
taosMemoryMalloc
(
tlen
);
if
(
buf
==
NULL
)
{
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
-
1
;
}
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
tlen
,
TD_ENCODER
);
t
EncoderInit
(
&
encoder
,
buf
,
tlen
);
tEncodeSMqCMCommitOffsetReq
(
&
encoder
,
&
req
);
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
pRequest
=
createRequest
(
tmq
->
pTscObj
,
NULL
,
NULL
,
TDMT_MND_MQ_COMMIT_OFFSET
);
if
(
pRequest
==
NULL
)
{
...
...
source/common/src/tmsg.c
浏览文件 @
323ce214
此差异已折叠。
点击以展开。
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @
323ce214
...
...
@@ -594,8 +594,8 @@ typedef struct {
SSchemaWrapper
outputSchema
;
}
SStreamObj
;
int32_t
tEncodeSStreamObj
(
S
C
oder
*
pEncoder
,
const
SStreamObj
*
pObj
);
int32_t
tDecodeSStreamObj
(
S
C
oder
*
pDecoder
,
SStreamObj
*
pObj
);
int32_t
tEncodeSStreamObj
(
S
Enc
oder
*
pEncoder
,
const
SStreamObj
*
pObj
);
int32_t
tDecodeSStreamObj
(
S
Dec
oder
*
pDecoder
,
SStreamObj
*
pObj
);
#ifdef __cplusplus
}
...
...
source/dnode/mnode/impl/src/mndDef.c
浏览文件 @
323ce214
...
...
@@ -411,7 +411,7 @@ void *tDecodeSMqSubActionLogObj(const void *buf, SMqSubActionLogObj *pLog) {
return
(
void
*
)
buf
;
}
int32_t
tEncodeSStreamObj
(
S
C
oder
*
pEncoder
,
const
SStreamObj
*
pObj
)
{
int32_t
tEncodeSStreamObj
(
S
Enc
oder
*
pEncoder
,
const
SStreamObj
*
pObj
)
{
int32_t
sz
=
0
;
/*int32_t outputNameSz = 0;*/
if
(
tEncodeCStr
(
pEncoder
,
pObj
->
name
)
<
0
)
return
-
1
;
...
...
@@ -462,7 +462,7 @@ int32_t tEncodeSStreamObj(SCoder *pEncoder, const SStreamObj *pObj) {
return
pEncoder
->
pos
;
}
int32_t
tDecodeSStreamObj
(
S
C
oder
*
pDecoder
,
SStreamObj
*
pObj
)
{
int32_t
tDecodeSStreamObj
(
S
Dec
oder
*
pDecoder
,
SStreamObj
*
pObj
)
{
if
(
tDecodeCStrTo
(
pDecoder
,
pObj
->
name
)
<
0
)
return
-
1
;
if
(
tDecodeCStrTo
(
pDecoder
,
pObj
->
sourceDb
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
pDecoder
,
&
pObj
->
createTime
)
<
0
)
return
-
1
;
...
...
source/dnode/mnode/impl/src/mndOffset.c
浏览文件 @
323ce214
...
...
@@ -157,8 +157,8 @@ static int32_t mndProcessCommitOffsetReq(SNodeMsg *pMsg) {
SMnode
*
pMnode
=
pMsg
->
pNode
;
char
*
msgStr
=
pMsg
->
rpcMsg
.
pCont
;
SMqCMCommitOffsetReq
commitOffsetReq
;
S
Coder
decoder
;
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
msgStr
,
pMsg
->
rpcMsg
.
contLen
,
TD_DECODER
);
S
Decoder
decoder
;
t
DecoderInit
(
&
decoder
,
msgStr
,
pMsg
->
rpcMsg
.
contLen
);
tDecodeSMqCMCommitOffsetReq
(
&
decoder
,
&
commitOffsetReq
);
...
...
source/dnode/mnode/impl/src/mndScheduler.c
浏览文件 @
323ce214
...
...
@@ -83,12 +83,12 @@ END:
}
int32_t
mndPersistTaskDeployReq
(
STrans
*
pTrans
,
SStreamTask
*
pTask
,
const
SEpSet
*
pEpSet
,
tmsg_t
type
,
int32_t
nodeId
)
{
S
C
oder
encoder
;
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
NULL
,
0
,
TD_ENCODER
);
S
Enc
oder
encoder
;
t
EncoderInit
(
&
encoder
,
NULL
,
0
);
tEncodeSStreamTask
(
&
encoder
,
pTask
);
int32_t
size
=
encoder
.
pos
;
int32_t
tlen
=
sizeof
(
SMsgHead
)
+
size
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
void
*
buf
=
taosMemoryMalloc
(
tlen
);
if
(
buf
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
...
...
@@ -96,9 +96,9 @@ int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet
}
((
SMsgHead
*
)
buf
)
->
vgId
=
htonl
(
nodeId
);
void
*
abuf
=
POINTER_SHIFT
(
buf
,
sizeof
(
SMsgHead
));
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
abuf
,
size
,
TD_ENCODER
);
t
EncoderInit
(
&
encoder
,
abuf
,
size
);
tEncodeSStreamTask
(
&
encoder
,
pTask
);
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
STransAction
action
=
{
0
};
memcpy
(
&
action
.
epSet
,
pEpSet
,
sizeof
(
SEpSet
));
...
...
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @
323ce214
...
...
@@ -369,7 +369,7 @@ static FORCE_INLINE int schemaExColIdCompare(const void *colId, const void *pSch
}
static
void
*
mndBuildVCreateStbReq
(
SMnode
*
pMnode
,
SVgObj
*
pVgroup
,
SStbObj
*
pStb
,
int32_t
*
pContLen
)
{
S
Coder
coder
=
{
0
};
S
Encoder
en
coder
=
{
0
};
int32_t
contLen
;
SName
name
=
{
0
};
SVCreateStbReq
req
=
{
0
};
...
...
@@ -422,11 +422,11 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
pHead
->
vgId
=
htonl
(
pVgroup
->
vgId
);
void
*
pBuf
=
POINTER_SHIFT
(
pHead
,
sizeof
(
SMsgHead
));
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pBuf
,
contLen
-
sizeof
(
SMsgHead
),
TD_ENCODER
);
if
(
tEncodeSVCreateStbReq
(
&
coder
,
&
req
)
<
0
)
{
t
EncoderInit
(
&
encoder
,
pBuf
,
contLen
-
sizeof
(
SMsgHead
)
);
if
(
tEncodeSVCreateStbReq
(
&
en
coder
,
&
req
)
<
0
)
{
return
NULL
;
}
t
CoderClear
(
&
coder
);
t
EncoderClear
(
&
en
coder
);
*
pContLen
=
contLen
;
taosMemoryFreeClear
(
req
.
pRSmaParam
.
qmsg1
);
...
...
@@ -440,7 +440,7 @@ static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb,
int32_t
contLen
=
0
;
int32_t
ret
=
0
;
SMsgHead
*
pHead
=
NULL
;
S
Coder
coder
=
{
0
};
S
Encoder
en
coder
=
{
0
};
tNameFromString
(
&
name
,
pStb
->
name
,
T_NAME_ACCT
|
T_NAME_DB
|
T_NAME_TABLE
);
...
...
@@ -462,9 +462,9 @@ static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb,
void
*
pBuf
=
POINTER_SHIFT
(
pHead
,
sizeof
(
SMsgHead
));
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pBuf
,
contLen
-
sizeof
(
SMsgHead
),
TD_ENCODER
);
tEncodeSVDropStbReq
(
&
coder
,
&
req
);
t
CoderClear
(
&
coder
);
t
EncoderInit
(
&
encoder
,
pBuf
,
contLen
-
sizeof
(
SMsgHead
)
);
tEncodeSVDropStbReq
(
&
en
coder
,
&
req
);
t
EncoderClear
(
&
en
coder
);
*
pContLen
=
contLen
;
return
pHead
;
...
...
source/dnode/mnode/impl/src/mndStream.c
浏览文件 @
323ce214
...
...
@@ -70,14 +70,14 @@ SSdbRaw *mndStreamActionEncode(SStreamObj *pStream) {
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
void
*
buf
=
NULL
;
S
C
oder
encoder
;
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
NULL
,
0
,
TD_ENCODER
);
S
Enc
oder
encoder
;
t
EncoderInit
(
&
encoder
,
NULL
,
0
);
if
(
tEncodeSStreamObj
(
&
encoder
,
pStream
)
<
0
)
{
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
goto
STREAM_ENCODE_OVER
;
}
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
int32_t
size
=
sizeof
(
int32_t
)
+
tlen
+
MND_STREAM_RESERVE_SIZE
;
SSdbRaw
*
pRaw
=
sdbAllocRaw
(
SDB_STREAM
,
MND_STREAM_VER_NUMBER
,
size
);
...
...
@@ -86,12 +86,12 @@ SSdbRaw *mndStreamActionEncode(SStreamObj *pStream) {
buf
=
taosMemoryMalloc
(
tlen
);
if
(
buf
==
NULL
)
goto
STREAM_ENCODE_OVER
;
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
tlen
,
TD_ENCODER
);
t
EncoderInit
(
&
encoder
,
buf
,
tlen
);
if
(
tEncodeSStreamObj
(
&
encoder
,
pStream
)
<
0
)
{
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
goto
STREAM_ENCODE_OVER
;
}
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
int32_t
dataPos
=
0
;
SDB_SET_INT32
(
pRaw
,
dataPos
,
tlen
,
STREAM_ENCODE_OVER
);
...
...
@@ -138,8 +138,8 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw) {
if
(
buf
==
NULL
)
goto
STREAM_DECODE_OVER
;
SDB_GET_BINARY
(
pRaw
,
dataPos
,
buf
,
tlen
,
STREAM_DECODE_OVER
);
S
C
oder
decoder
;
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
tlen
+
1
,
TD_DECODER
);
S
Dec
oder
decoder
;
t
DecoderInit
(
&
decoder
,
buf
,
tlen
+
1
);
if
(
tDecodeSStreamObj
(
&
decoder
,
pStream
)
<
0
)
{
goto
STREAM_DECODE_OVER
;
}
...
...
source/dnode/snode/src/snode.c
浏览文件 @
323ce214
...
...
@@ -99,10 +99,10 @@ void sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) {
ASSERT
(
0
);
return
;
}
S
C
oder
decoder
;
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
msg
,
pMsg
->
contLen
-
sizeof
(
SMsgHead
),
TD_DECODER
);
S
Dec
oder
decoder
;
t
DecoderInit
(
&
decoder
,
msg
,
pMsg
->
contLen
-
sizeof
(
SMsgHead
)
);
tDecodeSStreamTask
(
&
decoder
,
pTask
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
sndMetaDeployTask
(
pSnode
->
pMeta
,
pTask
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SND_TASK_EXEC
)
{
...
...
source/dnode/vnode/inc/vnode.h
浏览文件 @
323ce214
...
...
@@ -205,7 +205,7 @@ struct SMetaEntry {
struct
SMetaReader
{
int32_t
flags
;
SMeta
*
pMeta
;
S
Coder
coder
;
S
Decoder
coder
;
SMetaEntry
me
;
void
*
pBuf
;
int
szBuf
;
...
...
source/dnode/vnode/src/inc/meta.h
浏览文件 @
323ce214
...
...
@@ -39,8 +39,8 @@ typedef struct SMSmaCursor SMSmaCursor;
// metaOpen ==================
// metaEntry ==================
int
metaEncodeEntry
(
S
C
oder
*
pCoder
,
const
SMetaEntry
*
pME
);
int
metaDecodeEntry
(
S
C
oder
*
pCoder
,
SMetaEntry
*
pME
);
int
metaEncodeEntry
(
S
Enc
oder
*
pCoder
,
const
SMetaEntry
*
pME
);
int
metaDecodeEntry
(
S
Dec
oder
*
pCoder
,
SMetaEntry
*
pME
);
// metaTable ==================
...
...
source/dnode/vnode/src/meta/metaEntry.c
浏览文件 @
323ce214
...
...
@@ -15,7 +15,7 @@
#include "meta.h"
int
metaEncodeEntry
(
S
C
oder
*
pCoder
,
const
SMetaEntry
*
pME
)
{
int
metaEncodeEntry
(
S
Enc
oder
*
pCoder
,
const
SMetaEntry
*
pME
)
{
if
(
tStartEncode
(
pCoder
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
pCoder
,
pME
->
version
)
<
0
)
return
-
1
;
...
...
@@ -43,7 +43,7 @@ int metaEncodeEntry(SCoder *pCoder, const SMetaEntry *pME) {
return
0
;
}
int
metaDecodeEntry
(
S
C
oder
*
pCoder
,
SMetaEntry
*
pME
)
{
int
metaDecodeEntry
(
S
Dec
oder
*
pCoder
,
SMetaEntry
*
pME
)
{
uint32_t
len
;
if
(
tStartDecode
(
pCoder
)
<
0
)
return
-
1
;
...
...
source/dnode/vnode/src/meta/metaQuery.c
浏览文件 @
323ce214
...
...
@@ -22,7 +22,7 @@ void metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags) {
}
void
metaReaderClear
(
SMetaReader
*
pReader
)
{
t
C
oderClear
(
&
pReader
->
coder
);
t
Dec
oderClear
(
&
pReader
->
coder
);
tdbFree
(
pReader
->
pBuf
);
}
...
...
@@ -37,7 +37,7 @@ int metaGetTableEntryByVersion(SMetaReader *pReader, int64_t version, tb_uid_t u
}
// decode the entry
t
CoderInit
(
&
pReader
->
coder
,
TD_LITTLE_ENDIAN
,
pReader
->
pBuf
,
pReader
->
szBuf
,
TD_DECODER
);
t
DecoderInit
(
&
pReader
->
coder
,
pReader
->
pBuf
,
pReader
->
szBuf
);
if
(
metaDecodeEntry
(
&
pReader
->
coder
,
&
pReader
->
me
)
<
0
)
{
goto
_err
;
...
...
@@ -147,7 +147,7 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo
SSchemaWrapper
*
pSW
=
NULL
;
SSchema
*
pSchema
=
NULL
;
void
*
pBuf
;
S
Coder
coder
=
{
0
};
S
Decoder
coder
=
{
0
};
// fetch
skmDbKey
.
uid
=
uid
;
...
...
@@ -163,11 +163,11 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo
pBuf
=
pVal
;
pSW
=
taosMemoryMalloc
(
sizeof
(
SSchemaWrapper
));
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pVal
,
vLen
,
TD_DECODER
);
t
DecoderInit
(
&
coder
,
pVal
,
vLen
);
tDecodeSSchemaWrapper
(
&
coder
,
pSW
);
pSchema
=
taosMemoryMalloc
(
sizeof
(
SSchema
)
*
pSW
->
nCols
);
memcpy
(
pSchema
,
pSW
->
pSchema
,
sizeof
(
SSchema
)
*
pSW
->
nCols
);
t
C
oderClear
(
&
coder
);
t
Dec
oderClear
(
&
coder
);
pSW
->
pSchema
=
pSchema
;
...
...
source/dnode/vnode/src/meta/metaTable.c
浏览文件 @
323ce214
...
...
@@ -33,7 +33,6 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
void
*
pBuf
=
NULL
;
int32_t
szBuf
=
0
;
void
*
p
=
NULL
;
SCoder
coder
=
{
0
};
SMetaReader
mr
=
{
0
};
// validate req
...
...
@@ -192,7 +191,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
tb_uid_t
uid
;
int64_t
tver
;
SMetaEntry
me
=
{
0
};
S
Coder
coder
=
{
0
};
S
Decoder
coder
=
{
0
};
int8_t
type
;
int64_t
ctime
;
tb_uid_t
suid
;
...
...
@@ -253,7 +252,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
// decode entry
void
*
pDataCopy
=
taosMemoryMalloc
(
nData
);
// remove the copy (todo)
memcpy
(
pDataCopy
,
pData
,
nData
);
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pDataCopy
,
nData
,
TD_DECODER
);
t
DecoderInit
(
&
coder
,
pDataCopy
,
nData
);
ret
=
metaDecodeEntry
(
&
coder
,
&
me
);
if
(
ret
<
0
)
{
ASSERT
(
0
);
...
...
@@ -272,7 +271,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
}
taosMemoryFree
(
pDataCopy
);
t
C
oderClear
(
&
coder
);
t
Dec
oderClear
(
&
coder
);
tdbDbcClose
(
pTbDbc
);
if
(
type
==
TSDB_CHILD_TABLE
)
{
...
...
@@ -309,7 +308,7 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
void
*
pVal
=
NULL
;
int
kLen
=
0
;
int
vLen
=
0
;
S
Coder
coder
=
{
0
};
S
Encoder
coder
=
{
0
};
// set key and value
tbDbKey
.
version
=
pME
->
version
;
...
...
@@ -330,13 +329,13 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
goto
_err
;
}
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pVal
,
vLen
,
TD_ENCODER
);
t
EncoderInit
(
&
coder
,
pVal
,
vLen
);
if
(
metaEncodeEntry
(
&
coder
,
pME
)
<
0
)
{
goto
_err
;
}
t
C
oderClear
(
&
coder
);
t
Enc
oderClear
(
&
coder
);
// write to table.db
if
(
tdbDbInsert
(
pMeta
->
pTbDb
,
pKey
,
kLen
,
pVal
,
vLen
,
&
pMeta
->
txn
)
<
0
)
{
...
...
@@ -393,7 +392,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pME) {
}
static
int
metaSaveToSkmDb
(
SMeta
*
pMeta
,
const
SMetaEntry
*
pME
)
{
S
Coder
coder
=
{
0
};
S
Encoder
coder
=
{
0
};
void
*
pVal
=
NULL
;
int
vLen
=
0
;
int
rcode
=
0
;
...
...
@@ -422,7 +421,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
goto
_exit
;
}
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pVal
,
vLen
,
TD_ENCODER
);
t
EncoderInit
(
&
coder
,
pVal
,
vLen
);
tEncodeSSchemaWrapper
(
&
coder
,
pSW
);
if
(
tdbDbInsert
(
pMeta
->
pSkmDb
,
&
skmDbKey
,
sizeof
(
skmDbKey
),
pVal
,
vLen
,
&
pMeta
->
txn
)
<
0
)
{
...
...
@@ -432,7 +431,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
_exit:
taosMemoryFree
(
pVal
);
t
C
oderClear
(
&
coder
);
t
Enc
oderClear
(
&
coder
);
return
rcode
;
}
...
...
source/dnode/vnode/src/tq/tq.c
浏览文件 @
323ce214
...
...
@@ -910,12 +910,12 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
if
(
pTask
==
NULL
)
{
return
-
1
;
}
S
C
oder
decoder
;
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
(
uint8_t
*
)
msg
,
msgLen
,
TD_DECODER
);
S
Dec
oder
decoder
;
t
DecoderInit
(
&
decoder
,
(
uint8_t
*
)
msg
,
msgLen
);
if
(
tDecodeSStreamTask
(
&
decoder
,
pTask
)
<
0
)
{
ASSERT
(
0
);
}
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
// exec
if
(
tqExpandTask
(
pTq
,
pTask
,
4
)
<
0
)
{
...
...
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
浏览文件 @
323ce214
...
...
@@ -180,7 +180,7 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
const
uint8_t
*
pt
;
const
STSRow
*
pRow
;
uint64_t
szRow
;
S
Coder
coder
=
{
0
};
S
Decoder
de
coder
=
{
0
};
// tCoderInit(&coder, TD_LITTLE_ENDIAN, pSubmitBlk->pData, pSubmitBlk->nData, TD_DECODER);
for
(;;)
{
...
...
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @
323ce214
...
...
@@ -286,7 +286,7 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
static
int
vnodeProcessCreateStbReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int
len
,
SRpcMsg
*
pRsp
)
{
SVCreateStbReq
req
=
{
0
};
S
Coder
coder
;
S
Decoder
coder
;
pRsp
->
msgType
=
TDMT_VND_CREATE_STB_RSP
;
pRsp
->
code
=
TSDB_CODE_SUCCESS
;
...
...
@@ -294,7 +294,7 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq,
pRsp
->
contLen
=
0
;
// decode and process req
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pReq
,
len
,
TD_DECODER
);
t
DecoderInit
(
&
coder
,
pReq
,
len
);
if
(
tDecodeSVCreateStbReq
(
&
coder
,
&
req
)
<
0
)
{
pRsp
->
code
=
terrno
;
...
...
@@ -308,16 +308,16 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq,
tsdbRegisterRSma
(
pVnode
->
pTsdb
,
pVnode
->
pMeta
,
&
req
);
t
C
oderClear
(
&
coder
);
t
Dec
oderClear
(
&
coder
);
return
0
;
_err:
t
C
oderClear
(
&
coder
);
t
Dec
oderClear
(
&
coder
);
return
-
1
;
}
static
int
vnodeProcessCreateTbReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int
len
,
SRpcMsg
*
pRsp
)
{
S
Coder
coder
=
{
0
};
S
Decoder
de
coder
=
{
0
};
int
rcode
=
0
;
SVCreateTbBatchReq
req
=
{
0
};
SVCreateTbReq
*
pCreateReq
;
...
...
@@ -332,8 +332,8 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
pRsp
->
contLen
=
0
;
// decode
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pReq
,
len
,
TD_DECODER
);
if
(
tDecodeSVCreateTbBatchReq
(
&
coder
,
&
req
)
<
0
)
{
t
DecoderInit
(
&
decoder
,
pReq
,
len
);
if
(
tDecodeSVCreateTbBatchReq
(
&
de
coder
,
&
req
)
<
0
)
{
rcode
=
-
1
;
terrno
=
TSDB_CODE_INVALID_MSG
;
goto
_exit
;
...
...
@@ -373,13 +373,14 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
taosArrayPush
(
rsp
.
pArray
,
&
cRsp
);
}
t
CoderClear
(
&
coder
);
t
DecoderClear
(
&
de
coder
);
tsdbUpdateTbUidList
(
pVnode
->
pTsdb
,
pStore
);
tsdbUidStoreFree
(
pStore
);
// prepare rsp
int32_t
ret
=
0
;
SEncoder
encoder
=
{
0
};
int32_t
ret
=
0
;
tEncodeSize
(
tEncodeSVCreateTbBatchRsp
,
&
rsp
,
pRsp
->
contLen
,
ret
);
pRsp
->
pCont
=
rpcMallocCont
(
pRsp
->
contLen
);
if
(
pRsp
->
pCont
==
NULL
)
{
...
...
@@ -387,12 +388,14 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
rcode
=
-
1
;
goto
_exit
;
}
tCoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pRsp
->
pCont
,
pRsp
->
contLen
,
TD_ENCODER
);
tEncodeSVCreateTbBatchRsp
(
&
coder
,
&
rsp
);
tEncoderInit
(
&
encoder
,
pRsp
->
pCont
,
pRsp
->
contLen
);
tEncodeSVCreateTbBatchRsp
(
&
encoder
,
&
rsp
);
tEncoderClear
(
&
encoder
);
_exit:
taosArrayClear
(
rsp
.
pArray
);
tCoderClear
(
&
coder
);
tDecoderClear
(
&
decoder
);
tEncoderClear
(
&
encoder
);
return
rcode
;
}
...
...
@@ -416,15 +419,15 @@ static int vnodeProcessAlterStbReq(SVnode *pVnode, void *pReq, int32_t len, SRpc
static
int
vnodeProcessDropStbReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
)
{
SVDropStbReq
req
=
{
0
};
int
rcode
=
TSDB_CODE_SUCCESS
;
S
Coder
coder
=
{
0
};
S
Decoder
de
coder
=
{
0
};
pRsp
->
msgType
=
TDMT_VND_CREATE_STB_RSP
;
pRsp
->
pCont
=
NULL
;
pRsp
->
contLen
=
0
;
// decode request
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pReq
,
len
,
TD_DECODER
);
if
(
tDecodeSVDropStbReq
(
&
coder
,
&
req
)
<
0
)
{
t
DecoderInit
(
&
decoder
,
pReq
,
len
);
if
(
tDecodeSVDropStbReq
(
&
de
coder
,
&
req
)
<
0
)
{
rcode
=
TSDB_CODE_INVALID_MSG
;
goto
_exit
;
}
...
...
@@ -438,7 +441,7 @@ static int vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, i
// return rsp
_exit:
pRsp
->
code
=
rcode
;
t
CoderClear
(
&
coder
);
t
DecoderClear
(
&
de
coder
);
return
0
;
}
...
...
@@ -451,7 +454,7 @@ static int vnodeProcessAlterTbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcM
static
int
vnodeProcessDropTbReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
)
{
SVDropTbBatchReq
req
=
{
0
};
SVDropTbBatchRsp
rsp
=
{
0
};
S
Coder
coder
=
{
0
};
S
Decoder
de
coder
=
{
0
};
int
ret
;
pRsp
->
msgType
=
TDMT_VND_DROP_TABLE_RSP
;
...
...
@@ -460,8 +463,8 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
pRsp
->
code
=
TSDB_CODE_SUCCESS
;
// decode req
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pReq
,
len
,
TD_DECODER
);
ret
=
tDecodeSVDropTbBatchReq
(
&
coder
,
&
req
);
t
DecoderInit
(
&
decoder
,
pReq
,
len
);
ret
=
tDecodeSVDropTbBatchReq
(
&
de
coder
,
&
req
);
if
(
ret
<
0
)
{
terrno
=
TSDB_CODE_INVALID_MSG
;
pRsp
->
code
=
terrno
;
...
...
@@ -490,7 +493,7 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
}
_exit:
t
CoderClear
(
&
coder
);
t
DecoderClear
(
&
de
coder
);
// encode rsp (TODO)
return
0
;
}
...
...
@@ -501,7 +504,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
SSubmitBlk
*
pBlock
;
SSubmitRsp
rsp
=
{
0
};
SVCreateTbReq
createTbReq
=
{
0
};
S
Coder
coder
=
{
0
};
S
Decoder
de
coder
=
{
0
};
int32_t
nRows
;
pRsp
->
code
=
0
;
...
...
@@ -518,17 +521,17 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
// create table for auto create table mode
if
(
msgIter
.
schemaLen
>
0
)
{
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pBlock
->
data
,
msgIter
.
schemaLen
,
TD_DECODER
);
if
(
tDecodeSVCreateTbReq
(
&
coder
,
&
createTbReq
)
<
0
)
{
t
DecoderInit
(
&
decoder
,
pBlock
->
data
,
msgIter
.
schemaLen
);
if
(
tDecodeSVCreateTbReq
(
&
de
coder
,
&
createTbReq
)
<
0
)
{
pRsp
->
code
=
TSDB_CODE_INVALID_MSG
;
t
CoderClear
(
&
coder
);
t
DecoderClear
(
&
de
coder
);
goto
_exit
;
}
if
(
metaCreateTable
(
pVnode
->
pMeta
,
version
,
&
createTbReq
)
<
0
)
{
if
(
terrno
!=
TSDB_CODE_TDB_TABLE_ALREADY_EXIST
)
{
pRsp
->
code
=
terrno
;
t
CoderClear
(
&
coder
);
t
DecoderClear
(
&
de
coder
);
goto
_exit
;
}
}
...
...
@@ -540,7 +543,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
msgIter
.
suid
=
0
;
}
t
CoderClear
(
&
coder
);
t
DecoderClear
(
&
de
coder
);
}
if
(
tsdbInsertTableData
(
pVnode
->
pTsdb
,
&
msgIter
,
pBlock
,
&
nRows
)
<
0
)
{
...
...
source/libs/monitor/src/monMsg.c
浏览文件 @
323ce214
...
...
@@ -18,7 +18,7 @@
#include "tcoding.h"
#include "tencode.h"
static
int32_t
tEncodeSMonSysInfo
(
S
C
oder
*
encoder
,
const
SMonSysInfo
*
pInfo
)
{
static
int32_t
tEncodeSMonSysInfo
(
S
Enc
oder
*
encoder
,
const
SMonSysInfo
*
pInfo
)
{
if
(
tEncodeDouble
(
encoder
,
pInfo
->
cpu_engine
)
<
0
)
return
-
1
;
if
(
tEncodeDouble
(
encoder
,
pInfo
->
cpu_system
)
<
0
)
return
-
1
;
if
(
tEncodeFloat
(
encoder
,
pInfo
->
cpu_cores
)
<
0
)
return
-
1
;
...
...
@@ -37,7 +37,7 @@ static int32_t tEncodeSMonSysInfo(SCoder *encoder, const SMonSysInfo *pInfo) {
return
0
;
}
static
int32_t
tDecodeSMonSysInfo
(
S
C
oder
*
decoder
,
SMonSysInfo
*
pInfo
)
{
static
int32_t
tDecodeSMonSysInfo
(
S
Dec
oder
*
decoder
,
SMonSysInfo
*
pInfo
)
{
if
(
tDecodeDouble
(
decoder
,
&
pInfo
->
cpu_engine
)
<
0
)
return
-
1
;
if
(
tDecodeDouble
(
decoder
,
&
pInfo
->
cpu_system
)
<
0
)
return
-
1
;
if
(
tDecodeFloat
(
decoder
,
&
pInfo
->
cpu_cores
)
<
0
)
return
-
1
;
...
...
@@ -56,7 +56,7 @@ static int32_t tDecodeSMonSysInfo(SCoder *decoder, SMonSysInfo *pInfo) {
return
0
;
}
int32_t
tEncodeSMonLogs
(
S
C
oder
*
encoder
,
const
SMonLogs
*
pInfo
)
{
int32_t
tEncodeSMonLogs
(
S
Enc
oder
*
encoder
,
const
SMonLogs
*
pInfo
)
{
if
(
tEncodeI32
(
encoder
,
pInfo
->
numOfErrorLogs
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
encoder
,
pInfo
->
numOfInfoLogs
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
encoder
,
pInfo
->
numOfDebugLogs
)
<
0
)
return
-
1
;
...
...
@@ -71,7 +71,7 @@ int32_t tEncodeSMonLogs(SCoder *encoder, const SMonLogs *pInfo) {
return
0
;
}
static
int32_t
tDecodeSMonLogs
(
S
C
oder
*
decoder
,
SMonLogs
*
pInfo
)
{
static
int32_t
tDecodeSMonLogs
(
S
Dec
oder
*
decoder
,
SMonLogs
*
pInfo
)
{
if
(
tDecodeI32
(
decoder
,
&
pInfo
->
numOfErrorLogs
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
decoder
,
&
pInfo
->
numOfInfoLogs
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
decoder
,
&
pInfo
->
numOfDebugLogs
)
<
0
)
return
-
1
;
...
...
@@ -96,7 +96,7 @@ static int32_t tDecodeSMonLogs(SCoder *decoder, SMonLogs *pInfo) {
return
0
;
}
int32_t
tEncodeSMonClusterInfo
(
S
C
oder
*
encoder
,
const
SMonClusterInfo
*
pInfo
)
{
int32_t
tEncodeSMonClusterInfo
(
S
Enc
oder
*
encoder
,
const
SMonClusterInfo
*
pInfo
)
{
if
(
tEncodeCStr
(
encoder
,
pInfo
->
first_ep
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
encoder
,
pInfo
->
first_ep_dnode_id
)
<
0
)
return
-
1
;
if
(
tEncodeCStr
(
encoder
,
pInfo
->
version
)
<
0
)
return
-
1
;
...
...
@@ -124,7 +124,7 @@ int32_t tEncodeSMonClusterInfo(SCoder *encoder, const SMonClusterInfo *pInfo) {
return
0
;
}
int32_t
tDecodeSMonClusterInfo
(
S
C
oder
*
decoder
,
SMonClusterInfo
*
pInfo
)
{
int32_t
tDecodeSMonClusterInfo
(
S
Dec
oder
*
decoder
,
SMonClusterInfo
*
pInfo
)
{
if
(
tDecodeCStrTo
(
decoder
,
pInfo
->
first_ep
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
decoder
,
&
pInfo
->
first_ep_dnode_id
)
<
0
)
return
-
1
;
if
(
tDecodeCStrTo
(
decoder
,
pInfo
->
version
)
<
0
)
return
-
1
;
...
...
@@ -163,7 +163,7 @@ int32_t tDecodeSMonClusterInfo(SCoder *decoder, SMonClusterInfo *pInfo) {
return
0
;
}
int32_t
tEncodeSMonVgroupInfo
(
S
C
oder
*
encoder
,
const
SMonVgroupInfo
*
pInfo
)
{
int32_t
tEncodeSMonVgroupInfo
(
S
Enc
oder
*
encoder
,
const
SMonVgroupInfo
*
pInfo
)
{
if
(
tEncodeI32
(
encoder
,
taosArrayGetSize
(
pInfo
->
vgroups
))
<
0
)
return
-
1
;
for
(
int32_t
i
=
0
;
i
<
taosArrayGetSize
(
pInfo
->
vgroups
);
++
i
)
{
SMonVgroupDesc
*
pDesc
=
taosArrayGet
(
pInfo
->
vgroups
,
i
);
...
...
@@ -180,7 +180,7 @@ int32_t tEncodeSMonVgroupInfo(SCoder *encoder, const SMonVgroupInfo *pInfo) {
return
0
;
}
int32_t
tDecodeSMonVgroupInfo
(
S
C
oder
*
decoder
,
SMonVgroupInfo
*
pInfo
)
{
int32_t
tDecodeSMonVgroupInfo
(
S
Dec
oder
*
decoder
,
SMonVgroupInfo
*
pInfo
)
{
int32_t
arraySize
=
0
;
if
(
tDecodeI32
(
decoder
,
&
arraySize
)
<
0
)
return
-
1
;
...
...
@@ -203,14 +203,14 @@ int32_t tDecodeSMonVgroupInfo(SCoder *decoder, SMonVgroupInfo *pInfo) {
return
0
;
}
int32_t
tEncodeSMonGrantInfo
(
S
C
oder
*
encoder
,
const
SMonGrantInfo
*
pInfo
)
{
int32_t
tEncodeSMonGrantInfo
(
S
Enc
oder
*
encoder
,
const
SMonGrantInfo
*
pInfo
)
{
if
(
tEncodeI32
(
encoder
,
pInfo
->
expire_time
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
encoder
,
pInfo
->
timeseries_used
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
encoder
,
pInfo
->
timeseries_total
)
<
0
)
return
-
1
;
return
0
;
}
int32_t
tDecodeSMonGrantInfo
(
S
C
oder
*
decoder
,
SMonGrantInfo
*
pInfo
)
{
int32_t
tDecodeSMonGrantInfo
(
S
Dec
oder
*
decoder
,
SMonGrantInfo
*
pInfo
)
{
if
(
tDecodeI32
(
decoder
,
&
pInfo
->
expire_time
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
decoder
,
&
pInfo
->
timeseries_used
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
decoder
,
&
pInfo
->
timeseries_total
)
<
0
)
return
-
1
;
...
...
@@ -218,8 +218,8 @@ int32_t tDecodeSMonGrantInfo(SCoder *decoder, SMonGrantInfo *pInfo) {
}
int32_t
tSerializeSMonMmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonMmInfo
*
pInfo
)
{
S
C
oder
encoder
=
{
0
};
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
S
Enc
oder
encoder
=
{
0
};
t
EncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeSMonClusterInfo
(
&
encoder
,
&
pInfo
->
cluster
)
<
0
)
return
-
1
;
...
...
@@ -230,13 +230,13 @@ int32_t tSerializeSMonMmInfo(void *buf, int32_t bufLen, SMonMmInfo *pInfo) {
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
tlen
;
}
int32_t
tDeserializeSMonMmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonMmInfo
*
pInfo
)
{
S
C
oder
decoder
=
{
0
};
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
S
Dec
oder
decoder
=
{
0
};
t
DecoderInit
(
&
decoder
,
buf
,
bufLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tDecodeSMonClusterInfo
(
&
decoder
,
&
pInfo
->
cluster
)
<
0
)
return
-
1
;
...
...
@@ -246,7 +246,7 @@ int32_t tDeserializeSMonMmInfo(void *buf, int32_t bufLen, SMonMmInfo *pInfo) {
if
(
tDecodeSMonLogs
(
&
decoder
,
&
pInfo
->
log
)
<
0
)
return
-
1
;
tEndDecode
(
&
decoder
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
return
0
;
}
...
...
@@ -261,7 +261,7 @@ void tFreeSMonMmInfo(SMonMmInfo *pInfo) {
pInfo
->
log
.
logs
=
NULL
;
}
int32_t
tEncodeSMonDiskDesc
(
S
C
oder
*
encoder
,
const
SMonDiskDesc
*
pDesc
)
{
int32_t
tEncodeSMonDiskDesc
(
S
Enc
oder
*
encoder
,
const
SMonDiskDesc
*
pDesc
)
{
if
(
tEncodeCStr
(
encoder
,
pDesc
->
name
)
<
0
)
return
-
1
;
if
(
tEncodeI8
(
encoder
,
pDesc
->
level
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
encoder
,
pDesc
->
size
.
total
)
<
0
)
return
-
1
;
...
...
@@ -270,7 +270,7 @@ int32_t tEncodeSMonDiskDesc(SCoder *encoder, const SMonDiskDesc *pDesc) {
return
0
;
}
static
int32_t
tDecodeSMonDiskDesc
(
S
C
oder
*
decoder
,
SMonDiskDesc
*
pDesc
)
{
static
int32_t
tDecodeSMonDiskDesc
(
S
Dec
oder
*
decoder
,
SMonDiskDesc
*
pDesc
)
{
if
(
tDecodeCStrTo
(
decoder
,
pDesc
->
name
)
<
0
)
return
-
1
;
if
(
tDecodeI8
(
decoder
,
&
pDesc
->
level
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
decoder
,
&
pDesc
->
size
.
total
)
<
0
)
return
-
1
;
...
...
@@ -279,7 +279,7 @@ static int32_t tDecodeSMonDiskDesc(SCoder *decoder, SMonDiskDesc *pDesc) {
return
0
;
}
int32_t
tEncodeSMonDiskInfo
(
S
C
oder
*
encoder
,
const
SMonDiskInfo
*
pInfo
)
{
int32_t
tEncodeSMonDiskInfo
(
S
Enc
oder
*
encoder
,
const
SMonDiskInfo
*
pInfo
)
{
if
(
tEncodeI32
(
encoder
,
taosArrayGetSize
(
pInfo
->
datadirs
))
<
0
)
return
-
1
;
for
(
int32_t
i
=
0
;
i
<
taosArrayGetSize
(
pInfo
->
datadirs
);
++
i
)
{
SMonDiskDesc
*
pDesc
=
taosArrayGet
(
pInfo
->
datadirs
,
i
);
...
...
@@ -288,7 +288,7 @@ int32_t tEncodeSMonDiskInfo(SCoder *encoder, const SMonDiskInfo *pInfo) {
return
0
;
}
static
int32_t
tDecodeSMonDiskInfo
(
S
C
oder
*
decoder
,
SMonDiskInfo
*
pInfo
)
{
static
int32_t
tDecodeSMonDiskInfo
(
S
Dec
oder
*
decoder
,
SMonDiskInfo
*
pInfo
)
{
int32_t
arraySize
=
0
;
if
(
tDecodeI32
(
decoder
,
&
arraySize
)
<
0
)
return
-
1
;
...
...
@@ -304,7 +304,7 @@ static int32_t tDecodeSMonDiskInfo(SCoder *decoder, SMonDiskInfo *pInfo) {
return
0
;
}
int32_t
tEncodeSVnodesStat
(
S
C
oder
*
encoder
,
const
SVnodesStat
*
pStat
)
{
int32_t
tEncodeSVnodesStat
(
S
Enc
oder
*
encoder
,
const
SVnodesStat
*
pStat
)
{
if
(
tEncodeI32
(
encoder
,
pStat
->
openVnodes
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
encoder
,
pStat
->
totalVnodes
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
encoder
,
pStat
->
masterNum
)
<
0
)
return
-
1
;
...
...
@@ -317,7 +317,7 @@ int32_t tEncodeSVnodesStat(SCoder *encoder, const SVnodesStat *pStat) {
return
0
;
}
static
int32_t
tDecodeSVnodesStat
(
S
C
oder
*
decoder
,
SVnodesStat
*
pStat
)
{
static
int32_t
tDecodeSVnodesStat
(
S
Dec
oder
*
decoder
,
SVnodesStat
*
pStat
)
{
if
(
tDecodeI32
(
decoder
,
&
pStat
->
openVnodes
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
decoder
,
&
pStat
->
totalVnodes
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
decoder
,
&
pStat
->
masterNum
)
<
0
)
return
-
1
;
...
...
@@ -331,8 +331,8 @@ static int32_t tDecodeSVnodesStat(SCoder *decoder, SVnodesStat *pStat) {
}
int32_t
tSerializeSMonVmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonVmInfo
*
pInfo
)
{
S
C
oder
encoder
=
{
0
};
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
S
Enc
oder
encoder
=
{
0
};
t
EncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeSMonDiskInfo
(
&
encoder
,
&
pInfo
->
tfs
)
<
0
)
return
-
1
;
...
...
@@ -342,13 +342,13 @@ int32_t tSerializeSMonVmInfo(void *buf, int32_t bufLen, SMonVmInfo *pInfo) {
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
tlen
;
}
int32_t
tDeserializeSMonVmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonVmInfo
*
pInfo
)
{
S
C
oder
decoder
=
{
0
};
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
S
Dec
oder
decoder
=
{
0
};
t
DecoderInit
(
&
decoder
,
buf
,
bufLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tDecodeSMonDiskInfo
(
&
decoder
,
&
pInfo
->
tfs
)
<
0
)
return
-
1
;
...
...
@@ -357,7 +357,7 @@ int32_t tDeserializeSMonVmInfo(void *buf, int32_t bufLen, SMonVmInfo *pInfo) {
if
(
tDecodeSMonLogs
(
&
decoder
,
&
pInfo
->
log
)
<
0
)
return
-
1
;
tEndDecode
(
&
decoder
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
return
0
;
}
...
...
@@ -369,8 +369,8 @@ void tFreeSMonVmInfo(SMonVmInfo *pInfo) {
}
int32_t
tSerializeSMonQmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonQmInfo
*
pInfo
)
{
S
C
oder
encoder
=
{
0
};
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
S
Enc
oder
encoder
=
{
0
};
t
EncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeSMonSysInfo
(
&
encoder
,
&
pInfo
->
sys
)
<
0
)
return
-
1
;
...
...
@@ -378,20 +378,20 @@ int32_t tSerializeSMonQmInfo(void *buf, int32_t bufLen, SMonQmInfo *pInfo) {
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
tlen
;
}
int32_t
tDeserializeSMonQmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonQmInfo
*
pInfo
)
{
S
C
oder
decoder
=
{
0
};
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
S
Dec
oder
decoder
=
{
0
};
t
DecoderInit
(
&
decoder
,
buf
,
bufLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tDecodeSMonSysInfo
(
&
decoder
,
&
pInfo
->
sys
)
<
0
)
return
-
1
;
if
(
tDecodeSMonLogs
(
&
decoder
,
&
pInfo
->
log
)
<
0
)
return
-
1
;
tEndDecode
(
&
decoder
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
return
0
;
}
...
...
@@ -401,8 +401,8 @@ void tFreeSMonQmInfo(SMonQmInfo *pInfo) {
}
int32_t
tSerializeSMonSmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonSmInfo
*
pInfo
)
{
S
C
oder
encoder
=
{
0
};
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
S
Enc
oder
encoder
=
{
0
};
t
EncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeSMonSysInfo
(
&
encoder
,
&
pInfo
->
sys
)
<
0
)
return
-
1
;
...
...
@@ -410,20 +410,20 @@ int32_t tSerializeSMonSmInfo(void *buf, int32_t bufLen, SMonSmInfo *pInfo) {
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
tlen
;
}
int32_t
tDeserializeSMonSmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonSmInfo
*
pInfo
)
{
S
C
oder
decoder
=
{
0
};
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
S
Dec
oder
decoder
=
{
0
};
t
DecoderInit
(
&
decoder
,
buf
,
bufLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tDecodeSMonSysInfo
(
&
decoder
,
&
pInfo
->
sys
)
<
0
)
return
-
1
;
if
(
tDecodeSMonLogs
(
&
decoder
,
&
pInfo
->
log
)
<
0
)
return
-
1
;
tEndDecode
(
&
decoder
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
return
0
;
}
...
...
@@ -433,8 +433,8 @@ void tFreeSMonSmInfo(SMonSmInfo *pInfo) {
}
int32_t
tSerializeSMonBmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonBmInfo
*
pInfo
)
{
S
C
oder
encoder
=
{
0
};
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
S
Enc
oder
encoder
=
{
0
};
t
EncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeSMonSysInfo
(
&
encoder
,
&
pInfo
->
sys
)
<
0
)
return
-
1
;
...
...
@@ -442,20 +442,20 @@ int32_t tSerializeSMonBmInfo(void *buf, int32_t bufLen, SMonBmInfo *pInfo) {
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
tlen
;
}
int32_t
tDeserializeSMonBmInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonBmInfo
*
pInfo
)
{
S
C
oder
decoder
=
{
0
};
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
S
Dec
oder
decoder
=
{
0
};
t
DecoderInit
(
&
decoder
,
buf
,
bufLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tDecodeSMonSysInfo
(
&
decoder
,
&
pInfo
->
sys
)
<
0
)
return
-
1
;
if
(
tDecodeSMonLogs
(
&
decoder
,
&
pInfo
->
log
)
<
0
)
return
-
1
;
tEndDecode
(
&
decoder
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
return
0
;
}
...
...
@@ -465,8 +465,8 @@ void tFreeSMonBmInfo(SMonBmInfo *pInfo) {
}
int32_t
tSerializeSMonVloadInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonVloadInfo
*
pInfo
)
{
S
C
oder
encoder
=
{
0
};
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
S
Enc
oder
encoder
=
{
0
};
t
EncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
&
encoder
,
taosArrayGetSize
(
pInfo
->
pVloads
))
<
0
)
return
-
1
;
...
...
@@ -488,13 +488,13 @@ int32_t tSerializeSMonVloadInfo(void *buf, int32_t bufLen, SMonVloadInfo *pInfo)
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
tlen
;
}
int32_t
tDeserializeSMonVloadInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonVloadInfo
*
pInfo
)
{
S
C
oder
decoder
=
{
0
};
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
S
Dec
oder
decoder
=
{
0
};
t
DecoderInit
(
&
decoder
,
buf
,
bufLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
...
...
@@ -522,7 +522,7 @@ int32_t tDeserializeSMonVloadInfo(void *buf, int32_t bufLen, SMonVloadInfo *pInf
}
tEndDecode
(
&
decoder
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
return
0
;
}
...
...
@@ -532,8 +532,8 @@ void tFreeSMonVloadInfo(SMonVloadInfo *pInfo) {
}
int32_t
tSerializeSMonMloadInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonMloadInfo
*
pInfo
)
{
S
C
oder
encoder
=
{
0
};
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
S
Enc
oder
encoder
=
{
0
};
t
EncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeI8
(
&
encoder
,
pInfo
->
isMnode
)
<
0
)
return
-
1
;
...
...
@@ -541,19 +541,19 @@ int32_t tSerializeSMonMloadInfo(void *buf, int32_t bufLen, SMonMloadInfo *pInfo)
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
tlen
;
}
int32_t
tDeserializeSMonMloadInfo
(
void
*
buf
,
int32_t
bufLen
,
SMonMloadInfo
*
pInfo
)
{
S
C
oder
decoder
=
{
0
};
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
S
Dec
oder
decoder
=
{
0
};
t
DecoderInit
(
&
decoder
,
buf
,
bufLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tDecodeI8
(
&
decoder
,
&
pInfo
->
isMnode
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
&
decoder
,
&
pInfo
->
load
.
syncState
)
<
0
)
return
-
1
;
tEndDecode
(
&
decoder
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
return
0
;
}
\ No newline at end of file
source/libs/parser/src/parInsertData.c
浏览文件 @
323ce214
...
...
@@ -156,7 +156,7 @@ static int32_t createDataBlock(size_t defaultSize, int32_t rowSize, int32_t star
}
int32_t
buildCreateTbMsg
(
STableDataBlocks
*
pBlocks
,
SVCreateTbReq
*
pCreateTbReq
)
{
S
C
oder
coder
=
{
0
};
S
Enc
oder
coder
=
{
0
};
char
*
pBuf
;
int32_t
len
;
...
...
@@ -176,9 +176,9 @@ int32_t buildCreateTbMsg(STableDataBlocks* pBlocks, SVCreateTbReq* pCreateTbReq)
pBuf
=
pBlocks
->
pData
+
pBlocks
->
size
;
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pBuf
,
len
,
TD_ENCODER
);
t
EncoderInit
(
&
coder
,
pBuf
,
len
);
tEncodeSVCreateTbReq
(
&
coder
,
pCreateTbReq
);
t
C
oderClear
(
&
coder
);
t
Enc
oderClear
(
&
coder
);
pBlocks
->
size
+=
len
;
pBlocks
->
createTbReqLen
=
len
;
...
...
source/libs/parser/src/parTranslater.c
浏览文件 @
323ce214
...
...
@@ -3557,8 +3557,8 @@ static int32_t buildNormalTableBatchReq(int32_t acctId, const SCreateTableStmt*
}
static
int32_t
serializeVgroupCreateTableBatch
(
SVgroupCreateTableBatch
*
pTbBatch
,
SArray
*
pBufArray
)
{
int
tlen
;
S
C
oder
coder
=
{
0
};
int
tlen
;
S
Enc
oder
coder
=
{
0
};
int32_t
ret
=
0
;
tEncodeSize
(
tEncodeSVCreateTbBatchReq
,
&
pTbBatch
->
req
,
tlen
,
ret
);
...
...
@@ -3571,9 +3571,9 @@ static int32_t serializeVgroupCreateTableBatch(SVgroupCreateTableBatch* pTbBatch
((
SMsgHead
*
)
buf
)
->
contLen
=
htonl
(
tlen
);
void
*
pBuf
=
POINTER_SHIFT
(
buf
,
sizeof
(
SMsgHead
));
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pBuf
,
tlen
-
sizeof
(
SMsgHead
),
TD_ENCODER
);
t
EncoderInit
(
&
coder
,
pBuf
,
tlen
-
sizeof
(
SMsgHead
)
);
tEncodeSVCreateTbBatchReq
(
&
coder
,
&
pTbBatch
->
req
);
t
C
oderClear
(
&
coder
);
t
Enc
oderClear
(
&
coder
);
SVgDataBlocks
*
pVgData
=
taosMemoryCalloc
(
1
,
sizeof
(
SVgDataBlocks
));
if
(
NULL
==
pVgData
)
{
...
...
@@ -3957,8 +3957,8 @@ over:
static
void
destroyDropTbReqBatch
(
SVgroupDropTableBatch
*
pTbBatch
)
{
taosArrayDestroy
(
pTbBatch
->
req
.
pArray
);
}
static
int32_t
serializeVgroupDropTableBatch
(
SVgroupDropTableBatch
*
pTbBatch
,
SArray
*
pBufArray
)
{
int
tlen
;
S
C
oder
coder
=
{
0
};
int
tlen
;
S
Enc
oder
coder
=
{
0
};
int32_t
ret
=
0
;
tEncodeSize
(
tEncodeSVDropTbBatchReq
,
&
pTbBatch
->
req
,
tlen
,
ret
);
...
...
@@ -3971,9 +3971,9 @@ static int32_t serializeVgroupDropTableBatch(SVgroupDropTableBatch* pTbBatch, SA
((
SMsgHead
*
)
buf
)
->
contLen
=
htonl
(
tlen
);
void
*
pBuf
=
POINTER_SHIFT
(
buf
,
sizeof
(
SMsgHead
));
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pBuf
,
tlen
-
sizeof
(
SMsgHead
),
TD_ENCODER
);
t
EncoderInit
(
&
coder
,
pBuf
,
tlen
-
sizeof
(
SMsgHead
)
);
tEncodeSVDropTbBatchReq
(
&
coder
,
&
pTbBatch
->
req
);
t
C
oderClear
(
&
coder
);
t
Enc
oderClear
(
&
coder
);
SVgDataBlocks
*
pVgData
=
taosMemoryCalloc
(
1
,
sizeof
(
SVgDataBlocks
));
if
(
NULL
==
pVgData
)
{
...
...
source/libs/scheduler/src/scheduler.c
浏览文件 @
323ce214
...
...
@@ -1085,21 +1085,21 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
case
TDMT_VND_CREATE_TABLE_RSP
:
{
SVCreateTbBatchRsp
batchRsp
=
{
0
};
if
(
msg
)
{
S
C
oder
coder
=
{
0
};
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
msg
,
msgSize
,
TD_DECODER
);
S
Dec
oder
coder
=
{
0
};
t
DecoderInit
(
&
coder
,
msg
,
msgSize
);
code
=
tDecodeSVCreateTbBatchRsp
(
&
coder
,
&
batchRsp
);
if
(
TSDB_CODE_SUCCESS
==
code
&&
batchRsp
.
nRsps
>
0
)
{
for
(
int32_t
i
=
0
;
i
<
batchRsp
.
nRsps
;
++
i
)
{
SVCreateTbRsp
*
rsp
=
batchRsp
.
pRsps
+
i
;
if
(
NEED_CLIENT_HANDLE_ERROR
(
rsp
->
code
))
{
t
C
oderClear
(
&
coder
);
t
Dec
oderClear
(
&
coder
);
SCH_ERR_JRET
(
rsp
->
code
);
}
else
if
(
TSDB_CODE_SUCCESS
!=
rsp
->
code
)
{
code
=
rsp
->
code
;
}
}
}
t
C
oderClear
(
&
coder
);
t
Dec
oderClear
(
&
coder
);
SCH_ERR_JRET
(
code
);
}
...
...
@@ -1110,21 +1110,21 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
case
TDMT_VND_DROP_TABLE_RSP
:
{
SVDropTbBatchRsp
batchRsp
=
{
0
};
if
(
msg
)
{
S
C
oder
coder
=
{
0
};
t
CoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
msg
,
msgSize
,
TD_DECODER
);
S
Dec
oder
coder
=
{
0
};
t
DecoderInit
(
&
coder
,
msg
,
msgSize
);
code
=
tDecodeSVDropTbBatchRsp
(
&
coder
,
&
batchRsp
);
if
(
TSDB_CODE_SUCCESS
==
code
&&
batchRsp
.
nRsps
>
0
)
{
for
(
int32_t
i
=
0
;
i
<
batchRsp
.
nRsps
;
++
i
)
{
SVDropTbRsp
*
rsp
=
batchRsp
.
pRsps
+
i
;
if
(
NEED_CLIENT_HANDLE_ERROR
(
rsp
->
code
))
{
t
C
oderClear
(
&
coder
);
t
Dec
oderClear
(
&
coder
);
SCH_ERR_JRET
(
rsp
->
code
);
}
else
if
(
TSDB_CODE_SUCCESS
!=
rsp
->
code
)
{
code
=
rsp
->
code
;
}
}
}
t
C
oderClear
(
&
coder
);
t
Dec
oderClear
(
&
coder
);
SCH_ERR_JRET
(
code
);
}
...
...
source/libs/stream/src/tstream.c
浏览文件 @
323ce214
...
...
@@ -256,7 +256,7 @@ SStreamTask* tNewSStreamTask(int64_t streamId) {
return
pTask
;
}
int32_t
tEncodeSStreamTask
(
S
C
oder
*
pEncoder
,
const
SStreamTask
*
pTask
)
{
int32_t
tEncodeSStreamTask
(
S
Enc
oder
*
pEncoder
,
const
SStreamTask
*
pTask
)
{
/*if (tStartEncode(pEncoder) < 0) return -1;*/
if
(
tEncodeI64
(
pEncoder
,
pTask
->
streamId
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
pEncoder
,
pTask
->
taskId
)
<
0
)
return
-
1
;
...
...
@@ -301,7 +301,7 @@ int32_t tEncodeSStreamTask(SCoder* pEncoder, const SStreamTask* pTask) {
return
pEncoder
->
pos
;
}
int32_t
tDecodeSStreamTask
(
S
C
oder
*
pDecoder
,
SStreamTask
*
pTask
)
{
int32_t
tDecodeSStreamTask
(
S
Dec
oder
*
pDecoder
,
SStreamTask
*
pTask
)
{
/*if (tStartDecode(pDecoder) < 0) return -1;*/
if
(
tDecodeI64
(
pDecoder
,
&
pTask
->
streamId
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
pDecoder
,
&
pTask
->
taskId
)
<
0
)
return
-
1
;
...
...
source/libs/sync/src/syncMessage.c
浏览文件 @
323ce214
...
...
@@ -331,8 +331,8 @@ SyncPing* syncPingDeserialize2(const char* buf, uint32_t len) {
}
int32_t
syncPingSerialize3
(
const
SyncPing
*
pMsg
,
char
*
buf
,
int32_t
bufLen
)
{
S
C
oder
encoder
=
{
0
};
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
S
Enc
oder
encoder
=
{
0
};
t
EncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
{
return
-
1
;
}
...
...
@@ -367,13 +367,13 @@ int32_t syncPingSerialize3(const SyncPing* pMsg, char* buf, int32_t bufLen) {
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
tlen
;
}
SyncPing
*
syncPingDeserialize3
(
void
*
buf
,
int32_t
bufLen
)
{
S
C
oder
decoder
=
{
0
};
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
S
Dec
oder
decoder
=
{
0
};
t
DecoderInit
(
&
decoder
,
buf
,
bufLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
{
return
NULL
;
}
...
...
@@ -418,7 +418,7 @@ SyncPing* syncPingDeserialize3(void* buf, int32_t bufLen) {
memcpy
(
pMsg
->
data
,
data
,
len
);
tEndDecode
(
&
decoder
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
return
pMsg
;
}
...
...
@@ -590,8 +590,8 @@ SyncPingReply* syncPingReplyDeserialize2(const char* buf, uint32_t len) {
}
int32_t
syncPingReplySerialize3
(
const
SyncPingReply
*
pMsg
,
char
*
buf
,
int32_t
bufLen
)
{
S
C
oder
encoder
=
{
0
};
t
CoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
S
Enc
oder
encoder
=
{
0
};
t
EncoderInit
(
&
encoder
,
buf
,
bufLen
);
if
(
tStartEncode
(
&
encoder
)
<
0
)
{
return
-
1
;
}
...
...
@@ -626,13 +626,13 @@ int32_t syncPingReplySerialize3(const SyncPingReply* pMsg, char* buf, int32_t bu
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
t
C
oderClear
(
&
encoder
);
t
Enc
oderClear
(
&
encoder
);
return
tlen
;
}
SyncPingReply
*
syncPingReplyDeserialize3
(
void
*
buf
,
int32_t
bufLen
)
{
S
C
oder
decoder
=
{
0
};
t
CoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
S
Dec
oder
decoder
=
{
0
};
t
DecoderInit
(
&
decoder
,
buf
,
bufLen
);
if
(
tStartDecode
(
&
decoder
)
<
0
)
{
return
NULL
;
}
...
...
@@ -677,7 +677,7 @@ SyncPingReply* syncPingReplyDeserialize3(void* buf, int32_t bufLen) {
memcpy
(
pMsg
->
data
,
data
,
len
);
tEndDecode
(
&
decoder
);
t
C
oderClear
(
&
decoder
);
t
Dec
oderClear
(
&
decoder
);
return
pMsg
;
}
...
...
source/util/src/tencode.c
浏览文件 @
323ce214
...
...
@@ -21,48 +21,60 @@ static_assert(sizeof(float) == sizeof(uint32_t), "sizeof(float) must equal to si
static_assert
(
sizeof
(
double
)
==
sizeof
(
uint64_t
),
"sizeof(double) must equal to sizeof(uint64_t)"
);
#endif
void
tCoderInit
(
SCoder
*
pCoder
,
td_endian_t
endian
,
uint8_t
*
data
,
int32_t
size
,
td_coder_t
type
)
{
if
(
type
==
TD_ENCODER
)
{
if
(
data
==
NULL
)
size
=
0
;
}
else
{
ASSERT
(
data
&&
size
>
0
);
}
pCoder
->
type
=
type
;
pCoder
->
endian
=
endian
;
pCoder
->
data
=
data
;
pCoder
->
size
=
size
;
pCoder
->
pos
=
0
;
pCoder
->
mList
=
NULL
;
TD_SLIST_INIT
(
&
(
pCoder
->
stack
));
struct
SEncoderNode
{
SEncoderNode
*
pNext
;
uint8_t
*
data
;
uint32_t
size
;
uint32_t
pos
;
};
struct
SDecoderNode
{
SDecoderNode
*
pNext
;
const
uint8_t
*
data
;
uint32_t
size
;
uint32_t
pos
;
};
void
tEncoderInit
(
SEncoder
*
pEncoder
,
uint8_t
*
data
,
uint32_t
size
)
{
if
(
data
==
NULL
)
size
=
0
;
pEncoder
->
data
=
data
;
pEncoder
->
size
=
size
;
pEncoder
->
pos
=
0
;
pEncoder
->
mList
=
NULL
;
pEncoder
->
eStack
=
NULL
;
}
void
tCoderClear
(
SCoder
*
pCoder
)
{
SCoderMem
*
pMem
;
// clear memory
for
(
pMem
=
pCoder
->
mList
;
pMem
;
pMem
=
pCoder
->
mList
)
{
void
tEncoderClear
(
SEncoder
*
pCoder
)
{
for
(
SCoderMem
*
pMem
=
pCoder
->
mList
;
pMem
;
pMem
=
pCoder
->
mList
)
{
pCoder
->
mList
=
pMem
->
next
;
taosMemoryFree
(
pMem
);
}
memset
(
pCoder
,
0
,
sizeof
(
*
pCoder
));
}
void
tDecoderInit
(
SDecoder
*
pDecoder
,
const
uint8_t
*
data
,
uint32_t
size
)
{
pDecoder
->
data
=
data
;
pDecoder
->
size
=
size
;
pDecoder
->
pos
=
0
;
pDecoder
->
mList
=
NULL
;
pDecoder
->
dStack
=
NULL
;
}
struct
SCoderNode
*
pNode
;
for
(;;)
{
pNode
=
TD_SLIST_HEAD
(
&
(
pCoder
->
stack
));
if
(
pNode
==
NULL
)
break
;
TD_SLIST_POP
(
&
(
pCoder
->
stack
));
taosMemoryFree
(
pNode
);
void
tDecoderClear
(
SDecoder
*
pCoder
)
{
for
(
SCoderMem
*
pMem
=
pCoder
->
mList
;
pMem
;
pMem
=
pCoder
->
mList
)
{
pCoder
->
mList
=
pMem
->
next
;
taosMemoryFree
(
pMem
);
}
memset
(
pCoder
,
0
,
sizeof
(
*
pCoder
));
}
int32_t
tStartEncode
(
S
C
oder
*
pCoder
)
{
struct
SC
oderNode
*
pNode
;
int32_t
tStartEncode
(
S
Enc
oder
*
pCoder
)
{
SEnc
oderNode
*
pNode
;
ASSERT
(
pCoder
->
type
==
TD_ENCODER
);
if
(
pCoder
->
data
)
{
if
(
pCoder
->
size
-
pCoder
->
pos
<
sizeof
(
int32_t
))
return
-
1
;
pNode
=
t
aosMemoryMalloc
(
sizeof
(
*
pNode
));
pNode
=
t
EncoderMalloc
(
pCoder
,
sizeof
(
*
pNode
));
if
(
pNode
==
NULL
)
return
-
1
;
pNode
->
data
=
pCoder
->
data
;
...
...
@@ -73,22 +85,23 @@ int32_t tStartEncode(SCoder* pCoder) {
pCoder
->
pos
=
0
;
pCoder
->
size
=
pNode
->
size
-
pNode
->
pos
-
sizeof
(
int32_t
);
TD_SLIST_PUSH
(
&
(
pCoder
->
stack
),
pNode
);
pNode
->
pNext
=
pCoder
->
eStack
;
pCoder
->
eStack
=
pNode
;
}
else
{
pCoder
->
pos
+=
sizeof
(
int32_t
);
}
return
0
;
}
void
tEndEncode
(
S
C
oder
*
pCoder
)
{
struct
SC
oderNode
*
pNode
;
int32_t
len
;
void
tEndEncode
(
S
Enc
oder
*
pCoder
)
{
SEnc
oderNode
*
pNode
;
int32_t
len
;
ASSERT
(
pCoder
->
type
==
TD_ENCODER
);
if
(
pCoder
->
data
)
{
pNode
=
TD_SLIST_HEAD
(
&
(
pCoder
->
stack
))
;
pNode
=
pCoder
->
eStack
;
ASSERT
(
pNode
);
TD_SLIST_POP
(
&
(
pCoder
->
stack
))
;
pCoder
->
eStack
=
pNode
->
pNext
;
len
=
pCoder
->
pos
;
...
...
@@ -99,19 +112,16 @@ void tEndEncode(SCoder* pCoder) {
tEncodeI32
(
pCoder
,
len
);
TD_CODER_MOVE_POS
(
pCoder
,
len
);
taosMemoryFree
(
pNode
);
}
}
int32_t
tStartDecode
(
S
C
oder
*
pCoder
)
{
int32_t
len
;
struct
SCoderNode
*
pNode
;
int32_t
tStartDecode
(
S
Dec
oder
*
pCoder
)
{
SDecoderNode
*
pNode
;
int32_t
len
;
ASSERT
(
pCoder
->
type
==
TD_DECODER
);
if
(
tDecodeI32
(
pCoder
,
&
len
)
<
0
)
return
-
1
;
pNode
=
t
aosMemoryMalloc
(
sizeof
(
*
pNode
));
pNode
=
t
DecoderMalloc
(
pCoder
,
sizeof
(
*
pNode
));
if
(
pNode
==
NULL
)
return
-
1
;
pNode
->
data
=
pCoder
->
data
;
...
...
@@ -122,23 +132,20 @@ int32_t tStartDecode(SCoder* pCoder) {
pCoder
->
size
=
len
;
pCoder
->
pos
=
0
;
TD_SLIST_PUSH
(
&
(
pCoder
->
stack
),
pNode
);
pNode
->
pNext
=
pCoder
->
dStack
;
pCoder
->
dStack
=
pNode
;
return
0
;
}
void
tEndDecode
(
SCoder
*
pCoder
)
{
struct
SCoderNode
*
pNode
;
ASSERT
(
pCoder
->
type
==
TD_DECODER
);
void
tEndDecode
(
SDecoder
*
pCoder
)
{
SDecoderNode
*
pNode
;
pNode
=
TD_SLIST_HEAD
(
&
(
pCoder
->
stack
))
;
pNode
=
pCoder
->
dStack
;
ASSERT
(
pNode
);
TD_SLIST_POP
(
&
(
pCoder
->
stack
))
;
pCoder
->
dStack
=
pNode
->
pNext
;
pCoder
->
data
=
pNode
->
data
;
pCoder
->
pos
=
pCoder
->
size
+
pNode
->
pos
;
pCoder
->
size
=
pNode
->
size
;
taosMemoryFree
(
pNode
);
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录