Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Rattenking
TDengine
提交
1958981b
T
TDengine
项目概览
Rattenking
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
1958981b
编写于
4月 11, 2020
作者:
J
Jeff Tao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
add const in API
上级
068afbcd
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
66 addition
and
56 deletion
+66
-56
src/inc/trpc.h
src/inc/trpc.h
+5
-5
src/inc/tsync.h
src/inc/tsync.h
+25
-19
src/inc/twal.h
src/inc/twal.h
+10
-9
src/rpc/src/rpcMain.c
src/rpc/src/rpcMain.c
+12
-9
src/vnode/wal/src/walMain.c
src/vnode/wal/src/walMain.c
+14
-14
未找到文件。
src/inc/trpc.h
浏览文件 @
1958981b
...
...
@@ -76,16 +76,16 @@ typedef struct {
int
(
*
afp
)(
char
*
tableId
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
);
}
SRpcInit
;
void
*
rpcOpen
(
SRpcInit
*
pRpc
);
void
*
rpcOpen
(
const
SRpcInit
*
pRpc
);
void
rpcClose
(
void
*
);
void
*
rpcMallocCont
(
int
contLen
);
void
rpcFreeCont
(
void
*
pCont
);
void
*
rpcReallocCont
(
void
*
ptr
,
int
contLen
);
void
rpcSendRequest
(
void
*
thandle
,
SRpcIpSet
*
pIpSet
,
SRpcMsg
*
pMsg
);
void
rpcSendResponse
(
SRpcMsg
*
pMsg
);
void
rpcSendRedirectRsp
(
void
*
pConn
,
SRpcIpSet
*
pIpSet
);
void
rpcSendRequest
(
void
*
thandle
,
const
SRpcIpSet
*
pIpSet
,
const
SRpcMsg
*
pMsg
);
void
rpcSendResponse
(
const
SRpcMsg
*
pMsg
);
void
rpcSendRedirectRsp
(
void
*
pConn
,
const
SRpcIpSet
*
pIpSet
);
int
rpcGetConnInfo
(
void
*
thandle
,
SRpcConnInfo
*
pInfo
);
void
rpcSendRecv
(
void
*
shandle
,
SRpcIpSet
*
pIpSet
,
SRpcMsg
*
pOut
,
SRpcMsg
*
pRsp
);
void
rpcSendRecv
(
void
*
shandle
,
SRpcIpSet
*
pIpSet
,
const
SRpcMsg
*
pReq
,
SRpcMsg
*
pRsp
);
#ifdef __cplusplus
}
...
...
src/inc/tsync.h
浏览文件 @
1958981b
...
...
@@ -55,6 +55,24 @@ typedef struct {
int
role
[
TAOS_SYNC_MAX_REPLICA
];
}
SNodesRole
;
// if name is null, get the file from index or after, used by master
// if name is provided, get the named file at the specified index, used by unsynced node
// it returns the file magic number and size, if file not there, magic shall be 0.
typedef
uint32_t
(
*
FGetFileInfo
)(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
,
int32_t
*
size
);
// get the wal file from index or after
// return value, -1: error, 1:more wal files, 0:last WAL. if name[0]==0, no WAL file
typedef
int
(
*
FGetWalInfo
)(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
);
// when a forward pkt is received, call this to handle data
typedef
int
(
*
FWriteToCache
)(
void
*
ahandle
,
void
*
pHead
,
int
type
);
// when forward is confirmed by peer, master call this API to notify app
typedef
void
(
*
FConfirmForward
)(
void
*
ahandle
,
void
*
mhandle
,
int32_t
code
);
// when role is changed, call this to notify app
typedef
void
(
*
FNotifyRole
)(
void
*
ahandle
,
int8_t
role
);
typedef
struct
{
int32_t
vgId
;
// vgroup ID
uint64_t
version
;
// initial version
...
...
@@ -62,31 +80,19 @@ typedef struct {
char
path
[
128
];
// path to the file
void
*
ahandle
;
// handle provided by APP
FGetFileInfo
getFileInfo
;
FGetWalInfo
getWalInfo
;
FWriteToCache
writeToCache
;
FConfirmForward
confirmForward
;
FNotifyRole
notifyRole
;
// if name is null, get the file from index or after, used by master
// if name is provided, get the named file at the specified index, used by unsynced node
// it returns the file magic number and size, if file not there, magic shall be 0.
uint32_t
(
*
getFileInfo
)(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
,
int32_t
*
size
);
// get the wal file from index or after
// return value, -1: error, 1:more wal files, 0:last WAL. if name[0]==0, no WAL file
int
(
*
getWalInfo
)(
void
*
ahandle
,
char
*
name
,
uint32_t
*
index
);
// when a forward pkt is received, call this to handle data
int
(
*
writeToCache
)(
void
*
ahandle
,
void
*
pHead
,
int
type
);
// when forward is confirmed by peer, master call this API to notify app
void
(
*
confirmForward
)(
void
*
ahandle
,
void
*
mhandle
,
int32_t
code
);
// when role is changed, call this to notify app
void
(
*
notifyRole
)(
void
*
ahandle
,
int8_t
role
);
}
SSyncInfo
;
typedef
void
*
tsync_h
;
tsync_h
syncStart
(
SSyncInfo
*
);
tsync_h
syncStart
(
const
SSyncInfo
*
);
void
syncStop
(
tsync_h
shandle
);
int
syncReconfig
(
tsync_h
shandle
,
SSyncCfg
*
);
int
syncReconfig
(
tsync_h
shandle
,
const
SSyncCfg
*
);
int
syncForwardToPeer
(
tsync_h
shandle
,
void
*
pHead
,
void
*
mhandle
);
void
syncConfirmForward
(
tsync_h
shandle
,
uint64_t
version
,
int32_t
code
);
void
syncRecover
(
tsync_h
shandle
);
// recover from other nodes:
...
...
src/inc/twal.h
浏览文件 @
1958981b
...
...
@@ -38,15 +38,16 @@ typedef struct {
int8_t
wals
;
// number of WAL files;
}
SWalCfg
;
typedef
void
*
twal_h
;
// WAL HANDLE
twal_h
walOpen
(
char
*
path
,
SWalCfg
*
pCfg
);
void
walClose
(
twal_h
);
int
walRenew
(
twal_h
);
int
walWrite
(
twal_h
,
SWalHead
*
);
void
walFsync
(
twal_h
);
int
walRestore
(
twal_h
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
ahandle
,
void
*
pHead
,
int
type
));
int
walGetWalFile
(
twal_h
,
char
*
name
,
uint32_t
*
index
);
typedef
void
*
twalh
;
// WAL HANDLE
typedef
int
(
*
FWalWrite
)(
void
*
ahandle
,
void
*
pHead
,
int
type
);
twalh
walOpen
(
const
char
*
path
,
const
SWalCfg
*
pCfg
);
void
walClose
(
twalh
);
int
walRenew
(
twalh
);
int
walWrite
(
twalh
,
SWalHead
*
);
void
walFsync
(
twalh
);
int
walRestore
(
twalh
,
void
*
pVnode
,
FWalWrite
writeFp
);
int
walGetWalFile
(
twalh
,
char
*
name
,
uint32_t
*
index
);
extern
int
wDebugFlag
;
...
...
src/rpc/src/rpcMain.c
浏览文件 @
1958981b
...
...
@@ -202,7 +202,7 @@ static int rpcCheckAuthentication(SRpcConn *pConn, char *msg, int msgLen);
static
void
rpcLockConn
(
SRpcConn
*
pConn
);
static
void
rpcUnlockConn
(
SRpcConn
*
pConn
);
void
*
rpcOpen
(
SRpcInit
*
pInit
)
{
void
*
rpcOpen
(
const
SRpcInit
*
pInit
)
{
SRpcInfo
*
pRpc
;
tsRpcMaxRetry
=
tsRpcMaxTime
*
1000
/
tsRpcProgressTime
;
...
...
@@ -344,22 +344,22 @@ void *rpcReallocCont(void *ptr, int contLen) {
return
start
+
sizeof
(
SRpcReqContext
)
+
sizeof
(
SRpcHead
);
}
void
rpcSendRequest
(
void
*
shandle
,
SRpcIpSet
*
pIpSet
,
SRpcMsg
*
pMsg
)
{
void
rpcSendRequest
(
void
*
shandle
,
const
SRpcIpSet
*
pIpSet
,
const
SRpcMsg
*
pMsg
)
{
SRpcInfo
*
pRpc
=
(
SRpcInfo
*
)
shandle
;
SRpcReqContext
*
pContext
;
pMsg
->
contLen
=
rpcCompressRpcMsg
(
pMsg
->
pCont
,
pMsg
->
contLen
);
int
contLen
=
rpcCompressRpcMsg
(
pMsg
->
pCont
,
pMsg
->
contLen
);
pContext
=
(
SRpcReqContext
*
)
(
pMsg
->
pCont
-
sizeof
(
SRpcHead
)
-
sizeof
(
SRpcReqContext
));
pContext
->
ahandle
=
pMsg
->
handle
;
pContext
->
pRpc
=
(
SRpcInfo
*
)
shandle
;
pContext
->
ipSet
=
*
pIpSet
;
pContext
->
contLen
=
pMsg
->
contLen
;
pContext
->
contLen
=
contLen
;
pContext
->
pCont
=
pMsg
->
pCont
;
pContext
->
msgType
=
pMsg
->
msgType
;
pContext
->
oldInUse
=
pIpSet
->
inUse
;
pContext
->
connType
=
RPC_CONN_UDPC
;
if
(
pMsg
->
contLen
>
tsRpcMaxUdpSize
)
pContext
->
connType
=
RPC_CONN_TCPC
;
if
(
contLen
>
tsRpcMaxUdpSize
)
pContext
->
connType
=
RPC_CONN_TCPC
;
// connection type is application specific.
// for TDengine, all the query, show commands shall have TCP connection
...
...
@@ -374,11 +374,14 @@ void rpcSendRequest(void *shandle, SRpcIpSet *pIpSet, SRpcMsg *pMsg) {
return
;
}
void
rpcSendResponse
(
SRpcMsg
*
pMsg
)
{
void
rpcSendResponse
(
const
SRpcMsg
*
pRsp
)
{
int
msgLen
=
0
;
SRpcConn
*
pConn
=
(
SRpcConn
*
)
p
Msg
->
handle
;
SRpcConn
*
pConn
=
(
SRpcConn
*
)
p
Rsp
->
handle
;
SRpcInfo
*
pRpc
=
pConn
->
pRpc
;
SRpcMsg
rpcMsg
=
*
pRsp
;
SRpcMsg
*
pMsg
=
&
rpcMsg
;
if
(
pMsg
->
pCont
==
NULL
)
{
pMsg
->
pCont
=
rpcMallocCont
(
0
);
pMsg
->
contLen
=
0
;
...
...
@@ -429,7 +432,7 @@ void rpcSendResponse(SRpcMsg *pMsg) {
return
;
}
void
rpcSendRedirectRsp
(
void
*
thandle
,
SRpcIpSet
*
pIpSet
)
{
void
rpcSendRedirectRsp
(
void
*
thandle
,
const
SRpcIpSet
*
pIpSet
)
{
SRpcMsg
rpcMsg
;
rpcMsg
.
contLen
=
sizeof
(
SRpcIpSet
);
...
...
@@ -458,7 +461,7 @@ int rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
return
0
;
}
void
rpcSendRecv
(
void
*
shandle
,
SRpcIpSet
*
pIpSet
,
SRpcMsg
*
pMsg
,
SRpcMsg
*
pRsp
)
{
void
rpcSendRecv
(
void
*
shandle
,
SRpcIpSet
*
pIpSet
,
const
SRpcMsg
*
pMsg
,
SRpcMsg
*
pRsp
)
{
SRpcReqContext
*
pContext
;
pContext
=
(
SRpcReqContext
*
)
(
pMsg
->
pCont
-
sizeof
(
SRpcHead
)
-
sizeof
(
SRpcReqContext
));
...
...
src/vnode/wal/src/walMain.c
浏览文件 @
1958981b
...
...
@@ -48,11 +48,11 @@ typedef struct {
int
wDebugFlag
=
135
;
static
uint32_t
walSignature
=
0xFAFBFDFE
;
static
int
walHandleExistingFiles
(
char
*
path
);
static
int
walRestoreWalFile
(
c
har
*
name
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
,
void
*
,
int
)
);
static
int
walRemoveWalFiles
(
char
*
path
);
static
int
walHandleExistingFiles
(
c
onst
c
har
*
path
);
static
int
walRestoreWalFile
(
c
onst
char
*
name
,
void
*
pVnode
,
FWalWrite
writeFp
);
static
int
walRemoveWalFiles
(
c
onst
c
har
*
path
);
void
*
walOpen
(
c
har
*
path
,
SWalCfg
*
pCfg
)
{
void
*
walOpen
(
c
onst
char
*
path
,
const
SWalCfg
*
pCfg
)
{
SWal
*
pWal
=
calloc
(
sizeof
(
SWal
),
1
);
if
(
pWal
==
NULL
)
return
NULL
;
...
...
@@ -82,7 +82,7 @@ void *walOpen(char *path, SWalCfg *pCfg) {
void
walClose
(
void
*
handle
)
{
if
(
handle
==
NULL
)
return
;
SWal
*
pWal
=
(
SWal
*
)
handle
;
SWal
*
pWal
=
handle
;
close
(
pWal
->
fd
);
...
...
@@ -101,8 +101,8 @@ void walClose(void *handle) {
free
(
pWal
);
}
int
walRenew
(
twal_h
handle
)
{
SWal
*
pWal
=
(
SWal
*
)
handle
;
int
walRenew
(
void
*
handle
)
{
SWal
*
pWal
=
handle
;
int
code
=
0
;
pthread_mutex_lock
(
&
pWal
->
mutex
);
...
...
@@ -144,7 +144,7 @@ int walRenew(twal_h handle) {
}
int
walWrite
(
void
*
handle
,
SWalHead
*
pHead
)
{
SWal
*
pWal
=
(
SWal
*
)
handle
;
SWal
*
pWal
=
handle
;
int
code
=
0
;
// no wal
...
...
@@ -164,14 +164,14 @@ int walWrite(void *handle, SWalHead *pHead) {
void
walFsync
(
void
*
handle
)
{
SWal
*
pWal
=
(
SWal
*
)
handle
;
SWal
*
pWal
=
handle
;
if
(
pWal
->
level
==
TAOS_WAL_FSYNC
)
fsync
(
pWal
->
fd
);
}
int
walRestore
(
void
*
handle
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
,
void
*
,
int
))
{
SWal
*
pWal
=
(
SWal
*
)
handle
;
SWal
*
pWal
=
handle
;
int
code
=
0
;
struct
dirent
*
ent
;
int
count
=
0
;
...
...
@@ -223,7 +223,7 @@ int walRestore(void *handle, void *pVnode, int (*writeFp)(void *, void *, int))
}
int
walGetWalFile
(
void
*
handle
,
char
*
name
,
uint32_t
*
index
)
{
SWal
*
pWal
=
(
SWal
*
)
handle
;
SWal
*
pWal
=
handle
;
int
code
=
1
;
int32_t
first
=
0
;
...
...
@@ -247,7 +247,7 @@ int walGetWalFile(void *handle, char *name, uint32_t *index) {
return
code
;
}
static
int
walRestoreWalFile
(
c
har
*
name
,
void
*
pVnode
,
int
(
*
writeFp
)(
void
*
,
void
*
,
int
)
)
{
static
int
walRestoreWalFile
(
c
onst
char
*
name
,
void
*
pVnode
,
FWalWrite
writeFp
)
{
int
code
=
0
;
char
*
buffer
=
malloc
(
1024000
);
// size for one record
...
...
@@ -293,7 +293,7 @@ static int walRestoreWalFile(char *name, void *pVnode, int (*writeFp)(void *, vo
return
code
;
}
int
walHandleExistingFiles
(
char
*
path
)
{
int
walHandleExistingFiles
(
c
onst
c
har
*
path
)
{
int
code
=
0
;
char
oname
[
TSDB_FILENAME_LEN
*
3
];
char
nname
[
TSDB_FILENAME_LEN
*
3
];
...
...
@@ -335,7 +335,7 @@ int walHandleExistingFiles(char *path) {
return
code
;
}
static
int
walRemoveWalFiles
(
char
*
path
)
{
static
int
walRemoveWalFiles
(
c
onst
c
har
*
path
)
{
int
plen
=
strlen
(
walPrefix
);
char
name
[
TSDB_FILENAME_LEN
*
3
];
int
code
=
0
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录