Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
qq_50753323
TDengine
提交
f9598f47
T
TDengine
项目概览
qq_50753323
/
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,发现更多精彩内容 >>
未验证
提交
f9598f47
编写于
11月 18, 2020
作者:
S
Shengliang Guan
提交者:
GitHub
11月 18, 2020
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #4273 from taosdata/feature/wal
Feature/wal
上级
df74e517
52e99a8a
变更
8
显示空白变更内容
内联
并排
Showing
8 changed file
with
50 addition
and
38 deletion
+50
-38
src/dnode/inc/dnodeVRead.h
src/dnode/inc/dnodeVRead.h
+1
-1
src/dnode/src/dnodeVRead.c
src/dnode/src/dnodeVRead.c
+7
-9
src/inc/dnode.h
src/inc/dnode.h
+1
-1
src/inc/vnode.h
src/inc/vnode.h
+8
-7
src/vnode/inc/vnodeInt.h
src/vnode/inc/vnodeInt.h
+3
-2
src/vnode/src/vnodeMain.c
src/vnode/src/vnodeMain.c
+9
-8
src/vnode/src/vnodeRead.c
src/vnode/src/vnodeRead.c
+16
-5
src/vnode/src/vnodeWrite.c
src/vnode/src/vnodeWrite.c
+5
-5
未找到文件。
src/dnode/inc/dnodeVRead.h
浏览文件 @
f9598f47
...
...
@@ -24,7 +24,7 @@ int32_t dnodeInitVRead();
void
dnodeCleanupVRead
();
void
dnodeDispatchToVReadQueue
(
SRpcMsg
*
pMsg
);
void
*
dnodeAllocVReadQueue
(
void
*
pVnode
);
void
dnodeFreeVReadQueue
(
void
*
r
queue
);
void
dnodeFreeVReadQueue
(
void
*
pR
queue
);
#ifdef __cplusplus
}
...
...
src/dnode/src/dnodeVRead.c
浏览文件 @
f9598f47
...
...
@@ -35,7 +35,7 @@ typedef struct {
pthread_mutex_t
mutex
;
}
SVReadWorkerPool
;
static
void
*
dnodeProcessReadQueue
(
void
*
p
aram
);
static
void
*
dnodeProcessReadQueue
(
void
*
p
Worker
);
// module global variable
static
SVReadWorkerPool
tsVReadWP
;
...
...
@@ -47,7 +47,7 @@ int32_t dnodeInitVRead() {
tsVReadWP
.
min
=
tsNumOfCores
;
tsVReadWP
.
max
=
tsNumOfCores
*
tsNumOfThreadsPerCore
;
if
(
tsVReadWP
.
max
<=
tsVReadWP
.
min
*
2
)
tsVReadWP
.
max
=
2
*
tsVReadWP
.
min
;
tsVReadWP
.
worker
=
(
SVReadWorker
*
)
calloc
(
sizeof
(
SVReadWorker
),
tsVReadWP
.
max
);
tsVReadWP
.
worker
=
calloc
(
sizeof
(
SVReadWorker
),
tsVReadWP
.
max
);
pthread_mutex_init
(
&
tsVReadWP
.
mutex
,
NULL
);
if
(
tsVReadWP
.
worker
==
NULL
)
return
-
1
;
...
...
@@ -85,7 +85,7 @@ void dnodeCleanupVRead() {
void
dnodeDispatchToVReadQueue
(
SRpcMsg
*
pMsg
)
{
int32_t
queuedMsgNum
=
0
;
int32_t
leftLen
=
pMsg
->
contLen
;
char
*
pCont
=
(
char
*
)
pMsg
->
pCont
;
char
*
pCont
=
pMsg
->
pCont
;
while
(
leftLen
>
0
)
{
SMsgHead
*
pHead
=
(
SMsgHead
*
)
pCont
;
...
...
@@ -146,8 +146,8 @@ void *dnodeAllocVReadQueue(void *pVnode) {
return
queue
;
}
void
dnodeFreeVReadQueue
(
void
*
r
queue
)
{
taosCloseQueue
(
r
queue
);
void
dnodeFreeVReadQueue
(
void
*
pR
queue
)
{
taosCloseQueue
(
pR
queue
);
}
void
dnodeSendRpcVReadRsp
(
void
*
pVnode
,
SVReadMsg
*
pRead
,
int32_t
code
)
{
...
...
@@ -159,14 +159,12 @@ void dnodeSendRpcVReadRsp(void *pVnode, SVReadMsg *pRead, int32_t code) {
};
rpcSendResponse
(
&
rpcRsp
);
vnodeRelease
(
pVnode
);
}
void
dnodeDispatchNonRspMsg
(
void
*
pVnode
,
SVReadMsg
*
pRead
,
int32_t
code
)
{
vnodeRelease
(
pVnode
);
}
static
void
*
dnodeProcessReadQueue
(
void
*
p
aram
)
{
static
void
*
dnodeProcessReadQueue
(
void
*
p
Worker
)
{
SVReadMsg
*
pRead
;
int32_t
qtype
;
void
*
pVnode
;
...
...
@@ -193,7 +191,7 @@ static void *dnodeProcessReadQueue(void *param) {
}
}
taosFreeQitem
(
pRead
);
vnodeFreeFromRQueue
(
pVnode
,
pRead
);
}
return
NULL
;
...
...
src/inc/dnode.h
浏览文件 @
f9598f47
...
...
@@ -57,7 +57,7 @@ void *dnodeAllocVWriteQueue(void *pVnode);
void
dnodeFreeVWriteQueue
(
void
*
pWqueue
);
void
dnodeSendRpcVWriteRsp
(
void
*
pVnode
,
void
*
pWrite
,
int32_t
code
);
void
*
dnodeAllocVReadQueue
(
void
*
pVnode
);
void
dnodeFreeVReadQueue
(
void
*
r
queue
);
void
dnodeFreeVReadQueue
(
void
*
pR
queue
);
int32_t
dnodeAllocateMPeerQueue
();
void
dnodeFreeMPeerQueue
();
...
...
src/inc/vnode.h
浏览文件 @
f9598f47
...
...
@@ -23,12 +23,12 @@ extern "C" {
#include "twal.h"
typedef
enum
_VN_STATUS
{
TAOS_VN_STATUS_INIT
,
TAOS_VN_STATUS_READY
,
TAOS_VN_STATUS_CLOSING
,
TAOS_VN_STATUS_UPDATING
,
TAOS_VN_STATUS_RESET
,
}
EVnStatus
;
TAOS_VN_STATUS_INIT
=
0
,
TAOS_VN_STATUS_READY
=
1
,
TAOS_VN_STATUS_CLOSING
=
2
,
TAOS_VN_STATUS_UPDATING
=
3
,
TAOS_VN_STATUS_RESET
=
4
,
}
EVn
ode
Status
;
typedef
struct
{
int32_t
len
;
...
...
@@ -81,7 +81,8 @@ void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes);
int32_t
vnodeInitResources
();
void
vnodeCleanupResources
();
int32_t
vnodeWriteToRQueue
(
void
*
vparam
,
void
*
pCont
,
int32_t
contLen
,
int8_t
qtype
,
void
*
rparam
);
int32_t
vnodeWriteToRQueue
(
void
*
pVnode
,
void
*
pCont
,
int32_t
contLen
,
int8_t
qtype
,
void
*
rparam
);
void
vnodeFreeFromRQueue
(
void
*
pVnode
,
SVReadMsg
*
pRead
);
int32_t
vnodeProcessRead
(
void
*
pVnode
,
SVReadMsg
*
pRead
);
#ifdef __cplusplus
...
...
src/vnode/inc/vnodeInt.h
浏览文件 @
f9598f47
...
...
@@ -37,8 +37,9 @@ extern int32_t vDebugFlag;
typedef
struct
{
int32_t
vgId
;
// global vnode group ID
int32_t
refCount
;
// reference count
int32_t
queuedMsg
;
int32_t
delay
;
int32_t
queuedWMsg
;
int32_t
queuedRMsg
;
int32_t
delayMs
;
int8_t
status
;
int8_t
role
;
int8_t
accessState
;
...
...
src/vnode/src/vnodeMain.c
浏览文件 @
f9598f47
...
...
@@ -378,9 +378,9 @@ int32_t vnodeClose(int32_t vgId) {
return
0
;
}
void
vnodeRelease
(
void
*
pVnodeRaw
)
{
if
(
pVnodeRaw
==
NULL
)
return
;
SVnodeObj
*
pVnode
=
pVnodeRaw
;
void
vnodeRelease
(
void
*
vparam
)
{
if
(
vparam
==
NULL
)
return
;
SVnodeObj
*
pVnode
=
vparam
;
int32_t
code
=
0
;
int32_t
vgId
=
pVnode
->
vgId
;
...
...
@@ -643,18 +643,19 @@ static void vnodeNotifyRole(void *ahandle, int8_t role) {
pVnode
->
role
=
role
;
dnodeSendStatusMsgToMnode
();
if
(
pVnode
->
role
==
TAOS_SYNC_ROLE_MASTER
)
if
(
pVnode
->
role
==
TAOS_SYNC_ROLE_MASTER
)
{
cqStart
(
pVnode
->
cq
);
else
}
else
{
cqStop
(
pVnode
->
cq
);
}
}
static
void
vnodeCtrlFlow
(
void
*
ahandle
,
int32_t
mseconds
)
{
SVnodeObj
*
pVnode
=
ahandle
;
if
(
pVnode
->
delay
!=
mseconds
)
{
vInfo
(
"vgId:%d, sync flow control, mseconds:%d"
,
pVnode
->
vgId
,
mseconds
);
if
(
pVnode
->
delayMs
!=
mseconds
)
{
pVnode
->
delayMs
=
mseconds
;
vDebug
(
"vgId:%d, sync flow control, mseconds:%d"
,
pVnode
->
vgId
,
mseconds
);
}
pVnode
->
delay
=
mseconds
;
}
static
int
vnodeResetTsdb
(
SVnodeObj
*
pVnode
)
{
...
...
src/vnode/src/vnodeRead.c
浏览文件 @
f9598f47
...
...
@@ -41,8 +41,8 @@ void vnodeInitReadFp(void) {
// still required, or there will be a deadlock, so we don’t do any check here, but put the check codes before the
// request enters the queue
//
int32_t
vnodeProcessRead
(
void
*
param
,
SVReadMsg
*
pRead
)
{
SVnodeObj
*
pVnode
=
(
SVnodeObj
*
)
param
;
int32_t
vnodeProcessRead
(
void
*
v
param
,
SVReadMsg
*
pRead
)
{
SVnodeObj
*
pVnode
=
v
param
;
int32_t
msgType
=
pRead
->
msgType
;
if
(
vnodeProcessReadMsgFp
[
msgType
]
==
NULL
)
{
...
...
@@ -53,8 +53,8 @@ int32_t vnodeProcessRead(void *param, SVReadMsg *pRead) {
return
(
*
vnodeProcessReadMsgFp
[
msgType
])(
pVnode
,
pRead
);
}
static
int32_t
vnodeCheckRead
(
void
*
param
)
{
SVnodeObj
*
pVnode
=
param
;
static
int32_t
vnodeCheckRead
(
void
*
v
param
)
{
SVnodeObj
*
pVnode
=
v
param
;
if
(
pVnode
->
status
!=
TAOS_VN_STATUS_READY
)
{
vDebug
(
"vgId:%d, vnode status is %s, refCount:%d pVnode:%p"
,
pVnode
->
vgId
,
vnodeStatus
[
pVnode
->
status
],
pVnode
->
refCount
,
pVnode
);
...
...
@@ -76,6 +76,16 @@ static int32_t vnodeCheckRead(void *param) {
return
TSDB_CODE_SUCCESS
;
}
void
vnodeFreeFromRQueue
(
void
*
vparam
,
SVReadMsg
*
pRead
)
{
SVnodeObj
*
pVnode
=
vparam
;
atomic_sub_fetch_32
(
&
pVnode
->
queuedRMsg
,
1
);
vTrace
(
"vgId:%d, free from vrqueue, refCount:%d queued:%d"
,
pVnode
->
vgId
,
pVnode
->
refCount
,
pVnode
->
queuedRMsg
);
taosFreeQitem
(
pRead
);
vnodeRelease
(
pVnode
);
}
int32_t
vnodeWriteToRQueue
(
void
*
vparam
,
void
*
pCont
,
int32_t
contLen
,
int8_t
qtype
,
void
*
rparam
)
{
SVnodeObj
*
pVnode
=
vparam
;
...
...
@@ -108,7 +118,8 @@ int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qt
pRead
->
qtype
=
qtype
;
atomic_add_fetch_32
(
&
pVnode
->
refCount
,
1
);
vTrace
(
"vgId:%d, get vnode rqueue, refCount:%d pVnode:%p"
,
pVnode
->
vgId
,
pVnode
->
refCount
,
pVnode
);
atomic_add_fetch_32
(
&
pVnode
->
queuedRMsg
,
1
);
vTrace
(
"vgId:%d, write into vrqueue, refCount:%d queued:%d"
,
pVnode
->
vgId
,
pVnode
->
refCount
,
pVnode
->
queuedRMsg
);
taosWriteQitem
(
pVnode
->
rqueue
,
qtype
,
pRead
);
return
TSDB_CODE_SUCCESS
;
...
...
src/vnode/src/vnodeWrite.c
浏览文件 @
f9598f47
...
...
@@ -77,7 +77,7 @@ int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rpara
// assign version
pHead
->
version
=
pVnode
->
version
+
1
;
if
(
pVnode
->
delay
)
usleep
(
pVnode
->
delay
*
1000
);
if
(
pVnode
->
delay
Ms
)
taosMsleep
(
pVnode
->
delayMs
);
}
else
{
// from wal or forward
// for data from WAL or forward, version may be smaller
...
...
@@ -245,13 +245,13 @@ int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rpar
atomic_add_fetch_32
(
&
pVnode
->
refCount
,
1
);
int32_t
queued
=
atomic_add_fetch_32
(
&
pVnode
->
queuedMsg
,
1
);
int32_t
queued
=
atomic_add_fetch_32
(
&
pVnode
->
queued
W
Msg
,
1
);
if
(
queued
>
MAX_QUEUED_MSG_NUM
)
{
vDebug
(
"vgId:%d, too many msg:%d in vwqueue, flow control"
,
pVnode
->
vgId
,
queued
);
taosMsleep
(
1
);
}
vTrace
(
"vgId:%d, write into vwqueue, refCount:%d queued:%d"
,
pVnode
->
vgId
,
pVnode
->
refCount
,
pVnode
->
queuedMsg
);
vTrace
(
"vgId:%d, write into vwqueue, refCount:%d queued:%d"
,
pVnode
->
vgId
,
pVnode
->
refCount
,
pVnode
->
queued
W
Msg
);
taosWriteQitem
(
pVnode
->
wqueue
,
qtype
,
pWrite
);
return
TSDB_CODE_SUCCESS
;
...
...
@@ -260,8 +260,8 @@ int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rpar
void
vnodeFreeFromWQueue
(
void
*
vparam
,
SVWriteMsg
*
pWrite
)
{
SVnodeObj
*
pVnode
=
vparam
;
atomic_sub_fetch_32
(
&
pVnode
->
queuedMsg
,
1
);
vTrace
(
"vgId:%d, free from vwqueue, refCount:%d queued:%d"
,
pVnode
->
vgId
,
pVnode
->
refCount
,
pVnode
->
queuedMsg
);
atomic_sub_fetch_32
(
&
pVnode
->
queued
W
Msg
,
1
);
vTrace
(
"vgId:%d, free from vwqueue, refCount:%d queued:%d"
,
pVnode
->
vgId
,
pVnode
->
refCount
,
pVnode
->
queued
W
Msg
);
taosFreeQitem
(
pWrite
);
vnodeRelease
(
pVnode
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录