Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
2af3b3a2
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
2af3b3a2
编写于
1月 10, 2022
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
差异文件
Merge remote-tracking branch 'origin/3.0' into feature/dnode3
上级
155978b8
3cce6bd8
变更
17
显示空白变更内容
内联
并排
Showing
17 changed file
with
398 addition
and
156 deletion
+398
-156
include/common/tmsg.h
include/common/tmsg.h
+54
-34
include/libs/planner/planner.h
include/libs/planner/planner.h
+1
-0
include/libs/qcom/query.h
include/libs/qcom/query.h
+0
-49
include/libs/transport/trpc.h
include/libs/transport/trpc.h
+49
-0
include/util/thash.h
include/util/thash.h
+18
-0
source/client/inc/clientHb.h
source/client/inc/clientHb.h
+41
-16
source/client/src/clientHb.c
source/client/src/clientHb.c
+179
-29
source/common/src/tmsg.c
source/common/src/tmsg.c
+27
-16
source/dnode/mnode/impl/src/mndSync.c
source/dnode/mnode/impl/src/mndSync.c
+8
-7
source/libs/planner/CMakeLists.txt
source/libs/planner/CMakeLists.txt
+2
-1
source/libs/qcom/test/CMakeLists.txt
source/libs/qcom/test/CMakeLists.txt
+1
-1
source/libs/qcom/test/queryTest.cpp
source/libs/qcom/test/queryTest.cpp
+2
-1
source/libs/wal/inc/walInt.h
source/libs/wal/inc/walInt.h
+1
-0
source/libs/wal/src/walMeta.c
source/libs/wal/src/walMeta.c
+1
-0
source/libs/wal/src/walMgmt.c
source/libs/wal/src/walMgmt.c
+2
-1
source/libs/wal/src/walWrite.c
source/libs/wal/src/walWrite.c
+9
-0
source/util/src/thash.c
source/util/src/thash.c
+3
-1
未找到文件。
include/common/tmsg.h
浏览文件 @
2af3b3a2
...
...
@@ -133,47 +133,18 @@ typedef enum _mgmt_table {
#define TSDB_COL_IS_UD_COL(f) ((f & (~(TSDB_COL_NULL))) == TSDB_COL_UDC)
#define TSDB_COL_REQ_NULL(f) (((f)&TSDB_COL_NULL) != 0)
typedef
struct
SK
l
v
{
typedef
struct
SKv
{
int32_t
keyLen
;
int32_t
valueLen
;
void
*
key
;
void
*
value
;
}
SK
l
v
;
}
SKv
;
static
FORCE_INLINE
int
taosEncodeSKlv
(
void
**
buf
,
const
SKlv
*
pKlv
)
{
int
tlen
=
0
;
tlen
+=
taosEncodeFixedI32
(
buf
,
pKlv
->
keyLen
);
tlen
+=
taosEncodeFixedI32
(
buf
,
pKlv
->
valueLen
);
tlen
+=
taosEncodeBinary
(
buf
,
pKlv
->
key
,
pKlv
->
keyLen
);
tlen
+=
taosEncodeBinary
(
buf
,
pKlv
->
value
,
pKlv
->
valueLen
);
return
tlen
;
}
static
FORCE_INLINE
void
*
taosDecodeSKlv
(
void
*
buf
,
SKlv
*
pKlv
)
{
buf
=
taosDecodeFixedI32
(
buf
,
&
pKlv
->
keyLen
);
buf
=
taosDecodeFixedI32
(
buf
,
&
pKlv
->
valueLen
);
buf
=
taosDecodeBinary
(
buf
,
&
pKlv
->
key
,
pKlv
->
keyLen
);
buf
=
taosDecodeBinary
(
buf
,
&
pKlv
->
value
,
pKlv
->
valueLen
);
return
buf
;
}
typedef
struct
SClientHbKey
{
int32_t
connId
;
int32_t
hbType
;
}
SClientHbKey
;
static
FORCE_INLINE
int
taosEncodeSClientHbKey
(
void
**
buf
,
const
SClientHbKey
*
pKey
)
{
int
tlen
=
0
;
tlen
+=
taosEncodeFixedI32
(
buf
,
pKey
->
connId
);
tlen
+=
taosEncodeFixedI32
(
buf
,
pKey
->
hbType
);
return
tlen
;
}
static
FORCE_INLINE
void
*
taosDecodeSClientHbKey
(
void
*
buf
,
SClientHbKey
*
pKey
)
{
buf
=
taosDecodeFixedI32
(
buf
,
&
pKey
->
connId
);
buf
=
taosDecodeFixedI32
(
buf
,
&
pKey
->
hbType
);
return
buf
;
}
typedef
struct
SClientHbReq
{
SClientHbKey
connKey
;
SHashObj
*
info
;
// hash<Slv.key, Sklv>
...
...
@@ -184,9 +155,6 @@ typedef struct SClientHbBatchReq {
SArray
*
reqs
;
// SArray<SClientHbReq>
}
SClientHbBatchReq
;
int
tSerializeSClientHbReq
(
void
**
buf
,
const
SClientHbReq
*
pReq
);
void
*
tDeserializeClientHbReq
(
void
*
buf
,
SClientHbReq
*
pReq
);
typedef
struct
SClientHbRsp
{
SClientHbKey
connKey
;
int32_t
status
;
...
...
@@ -200,6 +168,58 @@ typedef struct SClientHbBatchRsp {
SArray
*
rsps
;
// SArray<SClientHbRsp>
}
SClientHbBatchRsp
;
static
FORCE_INLINE
uint32_t
hbKeyHashFunc
(
const
char
*
key
,
uint32_t
keyLen
)
{
return
taosIntHash_64
(
key
,
keyLen
);
}
int
tSerializeSClientHbReq
(
void
**
buf
,
const
SClientHbReq
*
pReq
);
void
*
tDeserializeClientHbReq
(
void
*
buf
,
SClientHbReq
*
pReq
);
static
FORCE_INLINE
void
tFreeClientHbReq
(
void
*
pReq
)
{
SClientHbReq
*
req
=
(
SClientHbReq
*
)
pReq
;
taosHashCleanup
(
req
->
info
);
free
(
pReq
);
}
int
tSerializeSClientHbBatchReq
(
void
**
buf
,
const
SClientHbBatchReq
*
pReq
);
void
*
tDeserializeClientHbBatchReq
(
void
*
buf
,
SClientHbBatchReq
*
pReq
);
static
FORCE_INLINE
void
tFreeClientHbBatchReq
(
void
*
pReq
)
{
SClientHbBatchReq
*
req
=
(
SClientHbBatchReq
*
)
pReq
;
taosArrayDestroyEx
(
req
->
reqs
,
tFreeClientHbReq
);
free
(
pReq
);
}
static
FORCE_INLINE
int
taosEncodeSKv
(
void
**
buf
,
const
SKv
*
pKv
)
{
int
tlen
=
0
;
tlen
+=
taosEncodeFixedI32
(
buf
,
pKv
->
keyLen
);
tlen
+=
taosEncodeFixedI32
(
buf
,
pKv
->
valueLen
);
tlen
+=
taosEncodeBinary
(
buf
,
pKv
->
key
,
pKv
->
keyLen
);
tlen
+=
taosEncodeBinary
(
buf
,
pKv
->
value
,
pKv
->
valueLen
);
return
tlen
;
}
static
FORCE_INLINE
void
*
taosDecodeSKv
(
void
*
buf
,
SKv
*
pKv
)
{
buf
=
taosDecodeFixedI32
(
buf
,
&
pKv
->
keyLen
);
buf
=
taosDecodeFixedI32
(
buf
,
&
pKv
->
valueLen
);
buf
=
taosDecodeBinary
(
buf
,
&
pKv
->
key
,
pKv
->
keyLen
);
buf
=
taosDecodeBinary
(
buf
,
&
pKv
->
value
,
pKv
->
valueLen
);
return
buf
;
}
static
FORCE_INLINE
int
taosEncodeSClientHbKey
(
void
**
buf
,
const
SClientHbKey
*
pKey
)
{
int
tlen
=
0
;
tlen
+=
taosEncodeFixedI32
(
buf
,
pKey
->
connId
);
tlen
+=
taosEncodeFixedI32
(
buf
,
pKey
->
hbType
);
return
tlen
;
}
static
FORCE_INLINE
void
*
taosDecodeSClientHbKey
(
void
*
buf
,
SClientHbKey
*
pKey
)
{
buf
=
taosDecodeFixedI32
(
buf
,
&
pKey
->
connId
);
buf
=
taosDecodeFixedI32
(
buf
,
&
pKey
->
hbType
);
return
buf
;
}
typedef
struct
SBuildTableMetaInput
{
int32_t
vgId
;
char
*
dbName
;
...
...
include/libs/planner/planner.h
浏览文件 @
2af3b3a2
...
...
@@ -23,6 +23,7 @@ extern "C" {
#include "query.h"
#include "tmsg.h"
#include "tarray.h"
#include "trpc.h"
#define QUERY_TYPE_MERGE 1
#define QUERY_TYPE_PARTIAL 2
...
...
include/libs/qcom/query.h
浏览文件 @
2af3b3a2
...
...
@@ -103,55 +103,6 @@ typedef struct STableMetaOutput {
STableMeta
*
tbMeta
;
}
STableMetaOutput
;
typedef
struct
SDataBuf
{
void
*
pData
;
uint32_t
len
;
}
SDataBuf
;
typedef
int32_t
(
*
__async_send_cb_fn_t
)(
void
*
param
,
const
SDataBuf
*
pMsg
,
int32_t
code
);
typedef
int32_t
(
*
__async_exec_fn_t
)(
void
*
param
);
typedef
struct
SMsgSendInfo
{
__async_send_cb_fn_t
fp
;
//async callback function
void
*
param
;
uint64_t
requestId
;
uint64_t
requestObjRefId
;
int32_t
msgType
;
SDataBuf
msgInfo
;
}
SMsgSendInfo
;
typedef
struct
SQueryNodeAddr
{
int32_t
nodeId
;
//vgId or qnodeId
int8_t
inUse
;
int8_t
numOfEps
;
SEpAddrMsg
epAddr
[
TSDB_MAX_REPLICA
];
}
SQueryNodeAddr
;
bool
tIsValidSchema
(
struct
SSchema
*
pSchema
,
int32_t
numOfCols
,
int32_t
numOfTags
);
int32_t
initTaskQueue
();
int32_t
cleanupTaskQueue
();
/**
*
* @param execFn The asynchronously execution function
* @param execParam The parameters of the execFn
* @param code The response code during execution the execFn
* @return
*/
int32_t
taosAsyncExec
(
__async_exec_fn_t
execFn
,
void
*
execParam
,
int32_t
*
code
);
/**
* Asynchronously send message to server, after the response received, the callback will be incured.
*
* @param pTransporter
* @param epSet
* @param pTransporterId
* @param pInfo
* @return
*/
int32_t
asyncSendMsgToServer
(
void
*
pTransporter
,
SEpSet
*
epSet
,
int64_t
*
pTransporterId
,
const
SMsgSendInfo
*
pInfo
);
const
SSchema
*
tGetTbnameColumnSchema
();
void
initQueryModuleMsgHandle
();
...
...
include/libs/transport/trpc.h
浏览文件 @
2af3b3a2
...
...
@@ -84,6 +84,55 @@ void rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp)
int
rpcReportProgress
(
void
*
pConn
,
char
*
pCont
,
int
contLen
);
void
rpcCancelRequest
(
int64_t
rid
);
typedef
struct
SDataBuf
{
void
*
pData
;
uint32_t
len
;
}
SDataBuf
;
typedef
int32_t
(
*
__async_send_cb_fn_t
)(
void
*
param
,
const
SDataBuf
*
pMsg
,
int32_t
code
);
typedef
int32_t
(
*
__async_exec_fn_t
)(
void
*
param
);
typedef
struct
SMsgSendInfo
{
__async_send_cb_fn_t
fp
;
//async callback function
void
*
param
;
uint64_t
requestId
;
uint64_t
requestObjRefId
;
int32_t
msgType
;
SDataBuf
msgInfo
;
}
SMsgSendInfo
;
typedef
struct
SQueryNodeAddr
{
int32_t
nodeId
;
//vgId or qnodeId
int8_t
inUse
;
int8_t
numOfEps
;
SEpAddrMsg
epAddr
[
TSDB_MAX_REPLICA
];
}
SQueryNodeAddr
;
bool
tIsValidSchema
(
struct
SSchema
*
pSchema
,
int32_t
numOfCols
,
int32_t
numOfTags
);
int32_t
initTaskQueue
();
int32_t
cleanupTaskQueue
();
/**
*
* @param execFn The asynchronously execution function
* @param execParam The parameters of the execFn
* @param code The response code during execution the execFn
* @return
*/
int32_t
taosAsyncExec
(
__async_exec_fn_t
execFn
,
void
*
execParam
,
int32_t
*
code
);
/**
* Asynchronously send message to server, after the response received, the callback will be incured.
*
* @param pTransporter
* @param epSet
* @param pTransporterId
* @param pInfo
* @return
*/
int32_t
asyncSendMsgToServer
(
void
*
pTransporter
,
SEpSet
*
epSet
,
int64_t
*
pTransporterId
,
const
SMsgSendInfo
*
pInfo
);
#ifdef __cplusplus
}
#endif
...
...
include/util/thash.h
浏览文件 @
2af3b3a2
...
...
@@ -214,6 +214,24 @@ void taosHashCancelIterate(SHashObj *pHashObj, void *p);
int32_t
taosHashGetKey
(
void
*
data
,
void
**
key
,
size_t
*
keyLen
);
/**
* Get the corresponding key information for a given data in hash table, using memcpy
* @param data
* @param dst
* @return
*/
static
FORCE_INLINE
int32_t
taosHashCopyKey
(
void
*
data
,
void
*
dst
)
{
if
(
NULL
==
data
||
NULL
==
dst
)
{
return
-
1
;
}
SHashNode
*
node
=
GET_HASH_PNODE
(
data
);
void
*
key
=
GET_HASH_NODE_KEY
(
node
);
memcpy
(
dst
,
key
,
node
->
keyLen
);
return
0
;
}
/**
* Get the corresponding data length for a given data in hash table
* @param data
...
...
source/client/inc/clientHb.h
浏览文件 @
2af3b3a2
...
...
@@ -18,36 +18,61 @@
#include "thash.h"
#include "tmsg.h"
#define HEARTBEAT_INTERVAL 1500 // ms
typedef
enum
{
mq
=
0
,
// type can be added here
HEARTBEAT_TYPE_MQ
=
0
,
// type
s
can be added here
//
HEARTBEAT_TYPE_MAX
}
EHbType
;
typedef
int32_t
(
*
FHbRspHandle
)(
SClientHbRsp
*
pReq
);
typedef
int32_t
(
*
FGetConnInfo
)(
SClientHbKey
connKey
,
void
*
param
);
typedef
struct
SClientHbMgr
{
int8_t
inited
;
int32_t
reportInterval
;
// unit ms
int32_t
stats
;
SRWLatch
lock
;
typedef
struct
SAppHbMgr
{
// statistics
int32_t
reportCnt
;
int32_t
connKeyCnt
;
int64_t
reportBytes
;
// not implemented
int64_t
startTime
;
// ctl
SRWLatch
lock
;
// lock is used in serialization
// connection
void
*
transporter
;
SEpSet
epSet
;
// info
SHashObj
*
activeInfo
;
// hash<SClientHbKey, SClientHbReq>
SHashObj
*
getInfoFuncs
;
// hash<SClientHbKey, FGetConnInfo>
}
SAppHbMgr
;
typedef
struct
SClientHbMgr
{
int8_t
inited
;
// ctl
int8_t
threadStop
;
pthread_t
thread
;
pthread_mutex_t
lock
;
// used when app init and cleanup
SArray
*
appHbMgrs
;
// SArray<SAppHbMgr*> one for each cluster
FHbRspHandle
handle
[
HEARTBEAT_TYPE_MAX
];
// input queue
}
SClientHbMgr
;
static
SClientHbMgr
clientHbMgr
=
{
0
};
// TODO: embed param into function
// return type: SArray<Skv>
typedef
SArray
*
(
*
FGetConnInfo
)(
SClientHbKey
connKey
,
void
*
param
);
// global, called by mgmt
int
hbMgrInit
();
void
hbMgrCleanUp
();
int
hbHandleRsp
(
void
*
hbMsg
);
int
hbHandleRsp
(
SClientHbBatchRsp
*
hbRsp
);
int
hbRegisterConn
(
SClientHbKey
connKey
,
FGetConnInfo
func
);
// cluster level
SAppHbMgr
*
appHbMgrInit
(
void
*
transporter
,
SEpSet
epSet
);
void
appHbMgrCleanup
(
SAppHbMgr
*
pAppHbMgr
);
// conn level
int
hbRegisterConn
(
SAppHbMgr
*
pAppHbMgr
,
SClientHbKey
connKey
,
FGetConnInfo
func
);
void
hbDeregisterConn
(
SAppHbMgr
*
pAppHbMgr
,
SClientHbKey
connKey
);
int
hbAddConnInfo
(
SClientHbKey
connKey
,
void
*
key
,
void
*
value
,
int32_t
keyLen
,
int32_t
valueLen
);
int
hbAddConnInfo
(
S
AppHbMgr
*
pAppHbMgr
,
S
ClientHbKey
connKey
,
void
*
key
,
void
*
value
,
int32_t
keyLen
,
int32_t
valueLen
);
// mq
void
hbMgrInitMqHbRspHandle
();
source/client/src/clientHb.c
浏览文件 @
2af3b3a2
...
...
@@ -14,64 +14,214 @@
*/
#include "clientHb.h"
#include "trpc.h"
static
int32_t
mqHbRspHandle
(
SClientHbRsp
*
pReq
)
{
static
SClientHbMgr
clientHbMgr
=
{
0
};
static
int32_t
hbCreateThread
();
static
void
hbStopThread
();
static
int32_t
hbMqHbRspHandle
(
SClientHbRsp
*
pReq
)
{
return
0
;
}
uint32_t
hbKeyHashFunc
(
const
char
*
key
,
uint32_t
keyLen
)
{
void
hbMgrInitMqHbRspHandle
()
{
clientHbMgr
.
handle
[
HEARTBEAT_TYPE_MQ
]
=
hbMqHbRspHandle
;
}
static
FORCE_INLINE
void
hbMgrInitHandle
()
{
// init all handle
hbMgrInitMqHbRspHandle
();
}
SClientHbBatchReq
*
hbGatherAllInfo
(
SAppHbMgr
*
pAppHbMgr
)
{
SClientHbBatchReq
*
pReq
=
malloc
(
sizeof
(
SClientHbBatchReq
));
if
(
pReq
==
NULL
)
{
terrno
=
TSDB_CODE_TSC_OUT_OF_MEMORY
;
return
NULL
;
}
int32_t
connKeyCnt
=
atomic_load_32
(
&
pAppHbMgr
->
connKeyCnt
);
pReq
->
reqs
=
taosArrayInit
(
connKeyCnt
,
sizeof
(
SClientHbReq
));
void
*
pIter
=
taosHashIterate
(
pAppHbMgr
->
activeInfo
,
NULL
);
while
(
pIter
!=
NULL
)
{
taosArrayPush
(
pReq
->
reqs
,
pIter
);
SClientHbReq
*
pOneReq
=
pIter
;
taosHashClear
(
pOneReq
->
info
);
pIter
=
taosHashIterate
(
pAppHbMgr
->
activeInfo
,
pIter
);
}
pIter
=
taosHashIterate
(
pAppHbMgr
->
getInfoFuncs
,
NULL
);
while
(
pIter
!=
NULL
)
{
FGetConnInfo
getConnInfoFp
=
(
FGetConnInfo
)
pIter
;
SClientHbKey
connKey
;
taosHashCopyKey
(
pIter
,
&
connKey
);
getConnInfoFp
(
connKey
,
NULL
);
pIter
=
taosHashIterate
(
pAppHbMgr
->
activeInfo
,
pIter
);
}
return
pReq
;
}
static
void
*
hbThreadFunc
(
void
*
param
)
{
setThreadName
(
"hb"
);
while
(
1
)
{
int8_t
threadStop
=
atomic_load_8
(
&
clientHbMgr
.
threadStop
);
if
(
threadStop
)
{
break
;
}
int
sz
=
taosArrayGetSize
(
clientHbMgr
.
appHbMgrs
);
for
(
int
i
=
0
;
i
<
sz
;
i
++
)
{
SAppHbMgr
*
pAppHbMgr
=
taosArrayGet
(
clientHbMgr
.
appHbMgrs
,
i
);
SClientHbBatchReq
*
pReq
=
hbGatherAllInfo
(
pAppHbMgr
);
void
*
reqStr
=
NULL
;
int
tlen
=
tSerializeSClientHbBatchReq
(
&
reqStr
,
pReq
);
SMsgSendInfo
info
;
/*info.fp = hbHandleRsp;*/
int64_t
transporterId
=
0
;
asyncSendMsgToServer
(
pAppHbMgr
->
transporter
,
&
pAppHbMgr
->
epSet
,
&
transporterId
,
&
info
);
tFreeClientHbBatchReq
(
pReq
);
atomic_add_fetch_32
(
&
pAppHbMgr
->
reportCnt
,
1
);
taosMsleep
(
HEARTBEAT_INTERVAL
);
}
}
return
NULL
;
}
static
int32_t
hbCreateThread
()
{
pthread_attr_t
thAttr
;
pthread_attr_init
(
&
thAttr
);
pthread_attr_setdetachstate
(
&
thAttr
,
PTHREAD_CREATE_JOINABLE
);
if
(
pthread_create
(
&
clientHbMgr
.
thread
,
&
thAttr
,
hbThreadFunc
,
NULL
)
!=
0
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
return
-
1
;
}
pthread_attr_destroy
(
&
thAttr
);
return
0
;
}
static
void
hbMgrInitMqHbFunc
()
{
clientHbMgr
.
handle
[
mq
]
=
mqHbRspHandle
;
static
void
hbStopThread
()
{
atomic_store_8
(
&
clientHbMgr
.
threadStop
,
1
);
}
SAppHbMgr
*
appHbMgrInit
(
void
*
transporter
,
SEpSet
epSet
)
{
SAppHbMgr
*
pAppHbMgr
=
malloc
(
sizeof
(
SAppHbMgr
));
if
(
pAppHbMgr
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
NULL
;
}
// init stat
pAppHbMgr
->
startTime
=
taosGetTimestampMs
();
// init connection info
pAppHbMgr
->
transporter
=
transporter
;
pAppHbMgr
->
epSet
=
epSet
;
// init hash info
pAppHbMgr
->
activeInfo
=
taosHashInit
(
64
,
hbKeyHashFunc
,
1
,
HASH_ENTRY_LOCK
);
pAppHbMgr
->
activeInfo
->
freeFp
=
tFreeClientHbReq
;
// init getInfoFunc
pAppHbMgr
->
getInfoFuncs
=
taosHashInit
(
64
,
hbKeyHashFunc
,
1
,
HASH_ENTRY_LOCK
);
taosArrayPush
(
clientHbMgr
.
appHbMgrs
,
&
pAppHbMgr
);
return
pAppHbMgr
;
}
void
appHbMgrCleanup
(
SAppHbMgr
*
pAppHbMgr
)
{
pthread_mutex_lock
(
&
clientHbMgr
.
lock
);
int
sz
=
taosArrayGetSize
(
clientHbMgr
.
appHbMgrs
);
for
(
int
i
=
0
;
i
<
sz
;
i
++
)
{
SAppHbMgr
*
pTarget
=
taosArrayGet
(
clientHbMgr
.
appHbMgrs
,
i
);
if
(
pAppHbMgr
==
pTarget
)
{
taosHashCleanup
(
pTarget
->
activeInfo
);
taosHashCleanup
(
pTarget
->
getInfoFuncs
);
}
}
pthread_mutex_unlock
(
&
clientHbMgr
.
lock
);
}
int
hbMgrInit
()
{
//init once
//
init once
int8_t
old
=
atomic_val_compare_exchange_8
(
&
clientHbMgr
.
inited
,
0
,
1
);
if
(
old
==
1
)
return
0
;
//init config
clientHbMgr
.
reportInterval
=
1500
;
//init stat
clientHbMgr
.
stats
=
0
;
clientHbMgr
.
appHbMgrs
=
taosArrayInit
(
0
,
sizeof
(
void
*
));
pthread_mutex_init
(
&
clientHbMgr
.
lock
,
NULL
);
//
init lock
taosInitRWLatch
(
&
clientHbMgr
.
lock
);
//
init handle funcs
hbMgrInitHandle
(
);
//
init handle funcs
hb
MgrInitMqHbFunc
();
//
init backgroud thread
hb
CreateThread
();
//init hash info
clientHbMgr
.
activeInfo
=
taosHashInit
(
64
,
hbKeyHashFunc
,
1
,
HASH_ENTRY_LOCK
);
//init getInfoFunc
clientHbMgr
.
getInfoFuncs
=
taosHashInit
(
64
,
hbKeyHashFunc
,
1
,
HASH_ENTRY_LOCK
);
return
0
;
}
void
hbMgrCleanUp
()
{
// destroy all appHbMgr
int8_t
old
=
atomic_val_compare_exchange_8
(
&
clientHbMgr
.
inited
,
1
,
0
);
if
(
old
==
0
)
return
;
taos
HashCleanup
(
clientHbMgr
.
activeInfo
);
taosHashCleanup
(
clientHbMgr
.
getInfoFuncs
);
taos
ArrayDestroy
(
clientHbMgr
.
appHbMgrs
);
}
int
hbRegisterConn
(
SClientHbKey
connKey
,
FGetConnInfo
func
)
{
int
hbHandleRsp
(
SClientHbBatchRsp
*
hbRsp
)
{
int64_t
reqId
=
hbRsp
->
reqId
;
int64_t
rspId
=
hbRsp
->
rspId
;
SArray
*
rsps
=
hbRsp
->
rsps
;
int32_t
sz
=
taosArrayGetSize
(
rsps
);
for
(
int
i
=
0
;
i
<
sz
;
i
++
)
{
SClientHbRsp
*
pRsp
=
taosArrayGet
(
rsps
,
i
);
if
(
pRsp
->
connKey
.
hbType
<
HEARTBEAT_TYPE_MAX
)
{
clientHbMgr
.
handle
[
pRsp
->
connKey
.
hbType
](
pRsp
);
}
else
{
// discard rsp
}
}
return
0
;
}
int
hbRegisterConn
(
SAppHbMgr
*
pAppHbMgr
,
SClientHbKey
connKey
,
FGetConnInfo
func
)
{
// init hash in activeinfo
void
*
data
=
taosHashGet
(
pAppHbMgr
->
activeInfo
,
&
connKey
,
sizeof
(
SClientHbKey
));
if
(
data
!=
NULL
)
{
return
0
;
}
SClientHbReq
hbReq
;
hbReq
.
connKey
=
connKey
;
hbReq
.
info
=
taosHashInit
(
64
,
hbKeyHashFunc
,
1
,
HASH_ENTRY_LOCK
);
taosHashPut
(
pAppHbMgr
->
activeInfo
,
&
connKey
,
sizeof
(
SClientHbKey
),
&
hbReq
,
sizeof
(
SClientHbReq
));
// init hash
if
(
func
!=
NULL
)
{
taosHashPut
(
pAppHbMgr
->
getInfoFuncs
,
&
connKey
,
sizeof
(
SClientHbKey
),
func
,
sizeof
(
FGetConnInfo
));
}
atomic_add_fetch_32
(
&
pAppHbMgr
->
connKeyCnt
,
1
);
return
0
;
}
void
hbDeregisterConn
(
SAppHbMgr
*
pAppHbMgr
,
SClientHbKey
connKey
)
{
taosHashRemove
(
pAppHbMgr
->
activeInfo
,
&
connKey
,
sizeof
(
SClientHbKey
));
taosHashRemove
(
pAppHbMgr
->
getInfoFuncs
,
&
connKey
,
sizeof
(
SClientHbKey
));
atomic_sub_fetch_32
(
&
pAppHbMgr
->
connKeyCnt
,
1
);
}
int
hbAddConnInfo
(
SClientHbKey
connKey
,
void
*
key
,
void
*
value
,
int32_t
keyLen
,
int32_t
valueLen
)
{
//lock
int
hbAddConnInfo
(
SAppHbMgr
*
pAppHbMgr
,
SClientHbKey
connKey
,
void
*
key
,
void
*
value
,
int32_t
keyLen
,
int32_t
valueLen
)
{
// find req by connection id
SClientHbReq
*
pReq
=
taosHashGet
(
pAppHbMgr
->
activeInfo
,
&
connKey
,
sizeof
(
SClientHbKey
));
ASSERT
(
pReq
!=
NULL
);
//find req by connection id
SClientHbReq
*
data
=
taosHashGet
(
clientHbMgr
.
activeInfo
,
&
connKey
,
sizeof
(
SClientHbKey
));
ASSERT
(
data
!=
NULL
);
taosHashPut
(
data
->
info
,
key
,
keyLen
,
value
,
valueLen
);
taosHashPut
(
pReq
->
info
,
key
,
keyLen
,
value
,
valueLen
);
//unlock
return
0
;
}
source/common/src/tmsg.c
浏览文件 @
2af3b3a2
...
...
@@ -89,17 +89,17 @@ int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) {
int
tlen
=
0
;
tlen
+=
taosEncodeSClientHbKey
(
buf
,
&
pReq
->
connKey
);
void
*
pIter
=
NULL
;
void
*
data
;
SK
lv
kl
v
;
data
=
taosHashIterate
(
pReq
->
info
,
pIter
);
while
(
data
!=
NULL
)
{
taosHashGetKey
(
data
,
&
klv
.
key
,
(
size_t
*
)
&
kl
v
.
keyLen
);
k
lv
.
valueLen
=
taosHashGetDataLen
(
data
);
k
lv
.
value
=
data
;
t
aosEncodeSKlv
(
buf
,
&
kl
v
);
data
=
taosHashIterate
(
pReq
->
info
,
pIter
);
int
kvNum
=
taosHashGetSize
(
pReq
->
info
)
;
tlen
+=
taosEncodeFixedI32
(
buf
,
kvNum
)
;
SK
v
k
v
;
void
*
pIter
=
taosHashIterate
(
pReq
->
info
,
pIter
);
while
(
pIter
!=
NULL
)
{
taosHashGetKey
(
pIter
,
&
kv
.
key
,
(
size_t
*
)
&
k
v
.
keyLen
);
k
v
.
valueLen
=
taosHashGetDataLen
(
pIter
);
k
v
.
value
=
pIter
;
t
len
+=
taosEncodeSKv
(
buf
,
&
k
v
);
pIter
=
taosHashIterate
(
pReq
->
info
,
pIter
);
}
return
tlen
;
}
...
...
@@ -109,16 +109,27 @@ void *tDeserializeClientHbReq(void *buf, SClientHbReq *pReq) {
buf
=
taosDecodeSClientHbKey
(
buf
,
&
pReq
->
connKey
);
// TODO: error handling
if
(
pReq
->
info
==
NULL
)
{
pReq
->
info
=
taosHashInit
(
64
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_BINARY
),
true
,
HASH_NO_LOCK
);
int
kvNum
;
taosDecodeFixedI32
(
buf
,
&
kvNum
);
pReq
->
info
=
taosHashInit
(
kvNum
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_BINARY
),
true
,
HASH_NO_LOCK
);
for
(
int
i
=
0
;
i
<
kvNum
;
i
++
)
{
SKv
kv
;
buf
=
taosDecodeSKv
(
buf
,
&
kv
);
taosHashPut
(
pReq
->
info
,
kv
.
key
,
kv
.
keyLen
,
kv
.
value
,
kv
.
valueLen
);
}
SKlv
klv
;
buf
=
taosDecodeSKlv
(
buf
,
&
klv
);
taosHashPut
(
pReq
->
info
,
klv
.
key
,
klv
.
keyLen
,
klv
.
value
,
klv
.
valueLen
);
return
buf
;
}
int
tSerializeSClientHbBatchReq
(
void
**
buf
,
const
SClientHbBatchReq
*
pReq
)
{
int
tlen
=
0
;
return
tlen
;
}
void
*
tDeserializeClientHbBatchReq
(
void
*
buf
,
SClientHbBatchReq
*
pReq
)
{
return
buf
;
}
int
tSerializeSVCreateTbReq
(
void
**
buf
,
SVCreateTbReq
*
pReq
)
{
int
tlen
=
0
;
...
...
source/dnode/mnode/impl/src/mndSync.c
浏览文件 @
2af3b3a2
...
...
@@ -91,7 +91,6 @@ static int32_t mndRestoreWal(SMnode *pMnode) {
if
(
sdbWriteFile
(
pSdb
)
!=
0
)
{
goto
WAL_RESTORE_OVER
;
}
}
if
(
walBeginSnapshot
(
pWal
,
sdbVer
)
<
0
)
{
goto
WAL_RESTORE_OVER
;
...
...
@@ -101,6 +100,8 @@ static int32_t mndRestoreWal(SMnode *pMnode) {
goto
WAL_RESTORE_OVER
;
}
}
code
=
0
;
WAL_RESTORE_OVER:
...
...
source/libs/planner/CMakeLists.txt
浏览文件 @
2af3b3a2
...
...
@@ -8,7 +8,8 @@ target_include_directories(
target_link_libraries
(
planner
PRIVATE os util catalog cjson parser transport function qcom
PRIVATE os util catalog cjson parser function qcom
PUBLIC transport
)
if
(
${
BUILD_TEST
}
)
...
...
source/libs/qcom/test/CMakeLists.txt
浏览文件 @
2af3b3a2
...
...
@@ -15,5 +15,5 @@ TARGET_INCLUDE_DIRECTORIES(
TARGET_LINK_LIBRARIES
(
queryUtilTest
PUBLIC os util gtest qcom common
PUBLIC os util gtest qcom common
transport
)
source/libs/qcom/test/queryTest.cpp
浏览文件 @
2af3b3a2
...
...
@@ -17,6 +17,7 @@
#include <iostream>
#include "tmsg.h"
#include "query.h"
#include "trpc.h"
#pragma GCC diagnostic ignored "-Wwrite-strings"
#pragma GCC diagnostic ignored "-Wunused-function"
...
...
source/libs/wal/inc/walInt.h
浏览文件 @
2af3b3a2
...
...
@@ -131,6 +131,7 @@ int walMetaDeserialize(SWal* pWal, const char* bytes);
// seek section
int
walChangeFile
(
SWal
*
pWal
,
int64_t
ver
);
int
walChangeFileToLast
(
SWal
*
pWal
);
// seek section end
int64_t
walGetSeq
();
...
...
source/libs/wal/src/walMeta.c
浏览文件 @
2af3b3a2
...
...
@@ -184,6 +184,7 @@ int walMetaDeserialize(SWal* pWal, const char* bytes) {
}
taosArraySetSize
(
pArray
,
sz
);
pWal
->
fileInfoSet
=
pArray
;
pWal
->
writeCur
=
sz
-
1
;
cJSON_Delete
(
pRoot
);
return
0
;
}
...
...
source/libs/wal/src/walMgmt.c
浏览文件 @
2af3b3a2
...
...
@@ -232,7 +232,8 @@ static int32_t walCreateThread() {
if
(
pthread_create
(
&
tsWal
.
thread
,
&
thAttr
,
walThreadFunc
,
NULL
)
!=
0
)
{
wError
(
"failed to create wal thread since %s"
,
strerror
(
errno
));
return
TAOS_SYSTEM_ERROR
(
errno
);
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
return
-
1
;
}
pthread_attr_destroy
(
&
thAttr
);
...
...
source/libs/wal/src/walWrite.c
浏览文件 @
2af3b3a2
...
...
@@ -244,6 +244,7 @@ int walRoll(SWal *pWal) {
pWal
->
writeIdxTfd
=
idxTfd
;
pWal
->
writeLogTfd
=
logTfd
;
pWal
->
writeCur
=
taosArrayGetSize
(
pWal
->
fileInfoSet
)
-
1
;
ASSERT
(
pWal
->
writeCur
>=
0
);
pWal
->
lastRollSeq
=
walGetSeq
();
return
0
;
...
...
@@ -253,6 +254,7 @@ static int walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) {
SWalIdxEntry
entry
=
{.
ver
=
ver
,
.
offset
=
offset
};
int
size
=
tfWrite
(
pWal
->
writeIdxTfd
,
&
entry
,
sizeof
(
SWalIdxEntry
));
if
(
size
!=
sizeof
(
SWalIdxEntry
))
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
// TODO truncate
return
-
1
;
}
...
...
@@ -286,7 +288,13 @@ int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, i
}
/*if (!tfValid(pWal->writeLogTfd)) return -1;*/
ASSERT
(
pWal
->
writeCur
>=
0
);
pthread_mutex_lock
(
&
pWal
->
mutex
);
if
(
pWal
->
writeIdxTfd
==
-
1
||
pWal
->
writeLogTfd
==
-
1
)
{
walChangeFileToLast
(
pWal
);
}
pWal
->
writeHead
.
head
.
version
=
index
;
int64_t
offset
=
walGetCurFileOffset
(
pWal
);
...
...
@@ -308,6 +316,7 @@ int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, i
wError
(
"vgId:%d, file:%"
PRId64
".log, failed to write since %s"
,
pWal
->
cfg
.
vgId
,
walGetLastFileFirstVer
(
pWal
),
strerror
(
errno
));
}
code
=
walWriteIndex
(
pWal
,
index
,
offset
);
if
(
code
!=
0
)
{
// TODO
...
...
source/util/src/thash.c
浏览文件 @
2af3b3a2
...
...
@@ -812,7 +812,9 @@ FORCE_INLINE int32_t taosHashGetKey(void *data, void** key, size_t* keyLen) {
SHashNode
*
node
=
GET_HASH_PNODE
(
data
);
*
key
=
GET_HASH_NODE_KEY
(
node
);
if
(
keyLen
)
{
*
keyLen
=
node
->
keyLen
;
}
return
0
;
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录