Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
a9aa8e58
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
Star
22014
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
a9aa8e58
编写于
5月 19, 2022
作者:
S
Shengliang Guan
提交者:
GitHub
5月 19, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #12697 from taosdata/fix/dnode
fix: acquire vnode on restart may deadlock
上级
fb81b902
2ee38b94
变更
15
隐藏空白更改
内联
并排
Showing
15 changed file
with
149 addition
and
142 deletion
+149
-142
include/dnode/mnode/sdb/sdb.h
include/dnode/mnode/sdb/sdb.h
+17
-17
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c
+4
-4
source/dnode/mgmt/mgmt_snode/inc/smInt.h
source/dnode/mgmt/mgmt_snode/inc/smInt.h
+0
-1
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
+15
-15
source/dnode/mgmt/mgmt_vnode/src/vmFile.c
source/dnode/mgmt/mgmt_vnode/src/vmFile.c
+2
-2
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
+2
-2
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+23
-18
source/dnode/mgmt/node_mgmt/inc/dmMgmt.h
source/dnode/mgmt/node_mgmt/inc/dmMgmt.h
+1
-1
source/dnode/mgmt/node_mgmt/src/dmMgmt.c
source/dnode/mgmt/node_mgmt/src/dmMgmt.c
+12
-10
source/dnode/mgmt/node_mgmt/src/dmNodes.c
source/dnode/mgmt/node_mgmt/src/dmNodes.c
+2
-2
source/dnode/mgmt/node_util/inc/dmUtil.h
source/dnode/mgmt/node_util/inc/dmUtil.h
+12
-12
source/dnode/mgmt/node_util/src/dmEps.c
source/dnode/mgmt/node_util/src/dmEps.c
+12
-13
source/dnode/mnode/sdb/src/sdb.c
source/dnode/mnode/sdb/src/sdb.c
+4
-2
source/dnode/mnode/sdb/src/sdbFile.c
source/dnode/mnode/sdb/src/sdbFile.c
+3
-3
source/dnode/mnode/sdb/src/sdbHash.c
source/dnode/mnode/sdb/src/sdbHash.c
+40
-40
未找到文件。
include/dnode/mnode/sdb/sdb.h
浏览文件 @
a9aa8e58
...
...
@@ -333,23 +333,23 @@ SSdbRow *sdbAllocRow(int32_t objSize);
void
*
sdbGetRowObj
(
SSdbRow
*
pRow
);
typedef
struct
SSdb
{
SMnode
*
pMnode
;
char
*
currDir
;
char
*
syncDir
;
char
*
tmpDir
;
int64_t
lastCommitVer
;
int64_t
curVer
;
int64_t
tableVer
[
SDB_MAX
];
int64_t
maxId
[
SDB_MAX
];
EKeyType
keyTypes
[
SDB_MAX
];
SHashObj
*
hashObjs
[
SDB_MAX
];
SRWLatch
locks
[
SDB_MAX
];
SdbInsertFp
insertFps
[
SDB_MAX
];
SdbUpdateFp
updateFps
[
SDB_MAX
];
SdbDeleteFp
deleteFps
[
SDB_MAX
];
SdbDeployFp
deployFps
[
SDB_MAX
];
SdbEncodeFp
encodeFps
[
SDB_MAX
];
SdbDecodeFp
decodeFps
[
SDB_MAX
];
SMnode
*
pMnode
;
char
*
currDir
;
char
*
syncDir
;
char
*
tmpDir
;
int64_t
lastCommitVer
;
int64_t
curVer
;
int64_t
tableVer
[
SDB_MAX
];
int64_t
maxId
[
SDB_MAX
];
EKeyType
keyTypes
[
SDB_MAX
];
SHashObj
*
hashObjs
[
SDB_MAX
];
TdThreadRwlock
locks
[
SDB_MAX
];
SdbInsertFp
insertFps
[
SDB_MAX
];
SdbUpdateFp
updateFps
[
SDB_MAX
];
SdbDeleteFp
deleteFps
[
SDB_MAX
];
SdbDeployFp
deployFps
[
SDB_MAX
];
SdbEncodeFp
encodeFps
[
SDB_MAX
];
SdbDecodeFp
decodeFps
[
SDB_MAX
];
}
SSdb
;
#ifdef __cplusplus
...
...
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c
浏览文件 @
a9aa8e58
...
...
@@ -19,11 +19,11 @@
static
void
dmUpdateDnodeCfg
(
SDnodeMgmt
*
pMgmt
,
SDnodeCfg
*
pCfg
)
{
if
(
pMgmt
->
pData
->
dnodeId
==
0
||
pMgmt
->
pData
->
clusterId
==
0
)
{
dInfo
(
"set dnodeId:%d clusterId:%"
PRId64
,
pCfg
->
dnodeId
,
pCfg
->
clusterId
);
taos
WLockLatch
(
&
pMgmt
->
pData
->
latch
);
taos
ThreadRwlockWrlock
(
&
pMgmt
->
pData
->
lock
);
pMgmt
->
pData
->
dnodeId
=
pCfg
->
dnodeId
;
pMgmt
->
pData
->
clusterId
=
pCfg
->
clusterId
;
dmWriteEps
(
pMgmt
->
pData
);
taos
WUnLockLatch
(
&
pMgmt
->
pData
->
latch
);
taos
ThreadRwlockUnlock
(
&
pMgmt
->
pData
->
lock
);
}
}
...
...
@@ -50,7 +50,7 @@ static void dmProcessStatusRsp(SDnodeMgmt *pMgmt, SRpcMsg *pRsp) {
void
dmSendStatusReq
(
SDnodeMgmt
*
pMgmt
)
{
SStatusReq
req
=
{
0
};
taos
RLockLatch
(
&
pMgmt
->
pData
->
latch
);
taos
ThreadRwlockRdlock
(
&
pMgmt
->
pData
->
lock
);
req
.
sver
=
tsVersion
;
req
.
dnodeVer
=
pMgmt
->
pData
->
dnodeVer
;
req
.
dnodeId
=
pMgmt
->
pData
->
dnodeId
;
...
...
@@ -69,7 +69,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
memcpy
(
req
.
clusterCfg
.
timezone
,
tsTimezoneStr
,
TD_TIMEZONE_LEN
);
memcpy
(
req
.
clusterCfg
.
locale
,
tsLocale
,
TD_LOCALE_LEN
);
memcpy
(
req
.
clusterCfg
.
charset
,
tsCharset
,
TD_LOCALE_LEN
);
taos
RUnLockLatch
(
&
pMgmt
->
pData
->
latch
);
taos
ThreadRwlockUnlock
(
&
pMgmt
->
pData
->
lock
);
SMonVloadInfo
vinfo
=
{
0
};
(
*
pMgmt
->
getVnodeLoadsFp
)(
&
vinfo
);
...
...
source/dnode/mgmt/mgmt_snode/inc/smInt.h
浏览文件 @
a9aa8e58
...
...
@@ -30,7 +30,6 @@ typedef struct SSnodeMgmt {
SMsgCb
msgCb
;
const
char
*
path
;
const
char
*
name
;
SRWLatch
latch
;
int8_t
uniqueWorkerInUse
;
SArray
*
uniqueWorkers
;
// SArray<SMultiWorker*>
SSingleWorker
sharedWorker
;
...
...
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
浏览文件 @
a9aa8e58
...
...
@@ -26,21 +26,21 @@ extern "C" {
#endif
typedef
struct
SVnodeMgmt
{
SDnodeData
*
pData
;
SMsgCb
msgCb
;
const
char
*
path
;
const
char
*
name
;
SQWorkerPool
queryPool
;
SQWorkerPool
fetchPool
;
SWWorkerPool
syncPool
;
SWWorkerPool
writePool
;
SWWorkerPool
mergePool
;
SSingleWorker
mgmtWorker
;
SSingleWorker
monitorWorker
;
SHashObj
*
hash
;
SRWLatch
latch
;
SVnodesStat
state
;
STfs
*
pTfs
;
SDnodeData
*
pData
;
SMsgCb
msgCb
;
const
char
*
path
;
const
char
*
name
;
SQWorkerPool
queryPool
;
SQWorkerPool
fetchPool
;
SWWorkerPool
syncPool
;
SWWorkerPool
writePool
;
SWWorkerPool
mergePool
;
SSingleWorker
mgmtWorker
;
SSingleWorker
monitorWorker
;
SHashObj
*
hash
;
TdThreadRwlock
lock
;
SVnodesStat
state
;
STfs
*
pTfs
;
}
SVnodeMgmt
;
typedef
struct
{
...
...
source/dnode/mgmt/mgmt_vnode/src/vmFile.c
浏览文件 @
a9aa8e58
...
...
@@ -17,7 +17,7 @@
#include "vmInt.h"
SVnodeObj
**
vmGetVnodeListFromHash
(
SVnodeMgmt
*
pMgmt
,
int32_t
*
numOfVnodes
)
{
taos
RLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockRdlock
(
&
pMgmt
->
lock
);
int32_t
num
=
0
;
int32_t
size
=
taosHashGetSize
(
pMgmt
->
hash
);
...
...
@@ -38,7 +38,7 @@ SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) {
}
}
taos
RUnLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockUnlock
(
&
pMgmt
->
lock
);
*
numOfVnodes
=
num
;
return
pVnodes
;
...
...
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @
a9aa8e58
...
...
@@ -20,7 +20,7 @@ void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
pInfo
->
pVloads
=
taosArrayInit
(
pMgmt
->
state
.
totalVnodes
,
sizeof
(
SVnodeLoad
));
if
(
pInfo
->
pVloads
==
NULL
)
return
;
taos
RLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockRdlock
(
&
pMgmt
->
lock
);
void
*
pIter
=
taosHashIterate
(
pMgmt
->
hash
,
NULL
);
while
(
pIter
)
{
...
...
@@ -34,7 +34,7 @@ void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
pIter
=
taosHashIterate
(
pMgmt
->
hash
,
pIter
);
}
taos
RUnLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockUnlock
(
&
pMgmt
->
lock
);
}
void
vmGetMonitorInfo
(
SVnodeMgmt
*
pMgmt
,
SMonVmInfo
*
pInfo
)
{
...
...
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
浏览文件 @
a9aa8e58
...
...
@@ -20,14 +20,14 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
SVnodeObj
*
pVnode
=
NULL
;
int32_t
refCount
=
0
;
taos
RLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockRdlock
(
&
pMgmt
->
lock
);
taosHashGetDup
(
pMgmt
->
hash
,
&
vgId
,
sizeof
(
int32_t
),
(
void
*
)
&
pVnode
);
if
(
pVnode
==
NULL
)
{
terrno
=
TSDB_CODE_VND_INVALID_VGROUP_ID
;
}
else
{
refCount
=
atomic_add_fetch_32
(
&
pVnode
->
refCount
,
1
);
}
taos
RUnLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockUnlock
(
&
pMgmt
->
lock
);
if
(
pVnode
!=
NULL
)
{
dTrace
(
"vgId:%d, acquire vnode, refCount:%d"
,
pVnode
->
vgId
,
refCount
);
...
...
@@ -39,9 +39,9 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
void
vmReleaseVnode
(
SVnodeMgmt
*
pMgmt
,
SVnodeObj
*
pVnode
)
{
if
(
pVnode
==
NULL
)
return
;
taos
RLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockRdlock
(
&
pMgmt
->
lock
);
int32_t
refCount
=
atomic_sub_fetch_32
(
&
pVnode
->
refCount
,
1
);
taos
RUnLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockUnlock
(
&
pMgmt
->
lock
);
dTrace
(
"vgId:%d, release vnode, refCount:%d"
,
pVnode
->
vgId
,
refCount
);
}
...
...
@@ -70,9 +70,9 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
return
-
1
;
}
taos
WLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockWrlock
(
&
pMgmt
->
lock
);
int32_t
code
=
taosHashPut
(
pMgmt
->
hash
,
&
pVnode
->
vgId
,
sizeof
(
int32_t
),
&
pVnode
,
sizeof
(
SVnodeObj
*
));
taos
WUnLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockUnlock
(
&
pMgmt
->
lock
);
return
code
;
}
...
...
@@ -80,9 +80,9 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
void
vmCloseVnode
(
SVnodeMgmt
*
pMgmt
,
SVnodeObj
*
pVnode
)
{
char
path
[
TSDB_FILENAME_LEN
]
=
{
0
};
taos
WLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockWrlock
(
&
pMgmt
->
lock
);
taosHashRemove
(
pMgmt
->
hash
,
&
pVnode
->
vgId
,
sizeof
(
int32_t
));
taos
WUnLockLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockUnlock
(
&
pMgmt
->
lock
);
vmReleaseVnode
(
pMgmt
,
pVnode
);
while
(
pVnode
->
refCount
>
0
)
taosMsleep
(
10
);
...
...
@@ -239,6 +239,7 @@ static void vmCleanup(SVnodeMgmt *pMgmt) {
vmStopWorker
(
pMgmt
);
vnodeCleanup
();
tfsClose
(
pMgmt
->
pTfs
);
taosThreadRwlockDestroy
(
&
pMgmt
->
lock
);
taosMemoryFree
(
pMgmt
);
}
...
...
@@ -260,7 +261,7 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
pMgmt
->
msgCb
.
queueFps
[
MERGE_QUEUE
]
=
(
PutToQueueFp
)
vmPutRpcMsgToMergeQueue
;
pMgmt
->
msgCb
.
qsizeFp
=
(
GetQueueSizeFp
)
vmGetQueueSize
;
pMgmt
->
msgCb
.
mgmt
=
pMgmt
;
taos
InitRWLatch
(
&
pMgmt
->
latch
);
taos
ThreadRwlockInit
(
&
pMgmt
->
lock
,
NULL
);
SDiskCfg
dCfg
=
{
0
};
tstrncpy
(
dCfg
.
dir
,
tsDataDir
,
TSDB_FILENAME_LEN
);
...
...
@@ -334,19 +335,23 @@ static int32_t vmRequire(const SMgmtInputOpt *pInput, bool *required) {
}
static
int32_t
vmStart
(
SVnodeMgmt
*
pMgmt
)
{
taosRLockLatch
(
&
pMgmt
->
latch
);
void
*
pIter
=
taosHashIterate
(
pMgmt
->
hash
,
NULL
);
while
(
pIter
)
{
SVnodeObj
**
ppVnode
=
pIter
;
if
(
ppVnode
==
NULL
||
*
ppVnode
==
NULL
)
continue
;
int32_t
numOfVnodes
=
0
;
SVnodeObj
**
pVnodes
=
vmGetVnodeListFromHash
(
pMgmt
,
&
numOfVnodes
);
SVnodeObj
*
pVnode
=
*
ppVnode
;
for
(
int32_t
i
=
0
;
i
<
numOfVnodes
;
++
i
)
{
SVnodeObj
*
pVnode
=
pVnodes
[
i
];
vnodeStart
(
pVnode
->
pImpl
);
pIter
=
taosHashIterate
(
pMgmt
->
hash
,
pIter
);
}
taosRUnLockLatch
(
&
pMgmt
->
latch
);
for
(
int32_t
i
=
0
;
i
<
numOfVnodes
;
++
i
)
{
SVnodeObj
*
pVnode
=
pVnodes
[
i
];
vmReleaseVnode
(
pMgmt
,
pVnode
);
}
if
(
pVnodes
!=
NULL
)
{
taosMemoryFree
(
pVnodes
);
}
return
0
;
}
...
...
source/dnode/mgmt/node_mgmt/inc/dmMgmt.h
浏览文件 @
a9aa8e58
...
...
@@ -70,7 +70,7 @@ typedef struct SMgmtWrapper {
const
char
*
name
;
char
*
path
;
int32_t
refCount
;
SRWLatch
latch
;
TdThreadRwlock
lock
;
EDndNodeType
ntype
;
bool
deployed
;
bool
required
;
...
...
source/dnode/mgmt/node_mgmt/src/dmMgmt.c
浏览文件 @
a9aa8e58
...
...
@@ -91,7 +91,7 @@ static int32_t dmInitVars(SDnode *pDnode, EDndNodeType rtype) {
return
-
1
;
}
taos
InitRWLatch
(
&
pData
->
latch
);
taos
ThreadRwlockInit
(
&
pData
->
lock
,
NULL
);
taosThreadMutexInit
(
&
pDnode
->
mutex
,
NULL
);
return
0
;
}
...
...
@@ -100,6 +100,7 @@ static void dmClearVars(SDnode *pDnode) {
for
(
EDndNodeType
ntype
=
DNODE
;
ntype
<
NODE_END
;
++
ntype
)
{
SMgmtWrapper
*
pWrapper
=
&
pDnode
->
wrappers
[
ntype
];
taosMemoryFreeClear
(
pWrapper
->
path
);
taosThreadRwlockDestroy
(
&
pWrapper
->
lock
);
}
if
(
pDnode
->
lockfile
!=
NULL
)
{
taosUnLockFile
(
pDnode
->
lockfile
);
...
...
@@ -108,7 +109,7 @@ static void dmClearVars(SDnode *pDnode) {
}
SDnodeData
*
pData
=
&
pDnode
->
data
;
taos
WLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockWrlock
(
&
pData
->
lock
);
if
(
pData
->
dnodeEps
!=
NULL
)
{
taosArrayDestroy
(
pData
->
dnodeEps
);
pData
->
dnodeEps
=
NULL
;
...
...
@@ -117,8 +118,9 @@ static void dmClearVars(SDnode *pDnode) {
taosHashCleanup
(
pData
->
dnodeHash
);
pData
->
dnodeHash
=
NULL
;
}
taos
WUnLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockUnlock
(
&
pData
->
lock
);
taosThreadRwlockDestroy
(
&
pData
->
lock
);
taosThreadMutexDestroy
(
&
pDnode
->
mutex
);
memset
(
&
pDnode
->
mutex
,
0
,
sizeof
(
pDnode
->
mutex
));
}
...
...
@@ -151,7 +153,7 @@ int32_t dmInitDnode(SDnode *pDnode, EDndNodeType rtype) {
if
(
ntype
==
DNODE
)
{
pWrapper
->
proc
.
ptype
=
DND_PROC_SINGLE
;
}
taos
InitRWLatch
(
&
pWrapper
->
latch
);
taos
ThreadRwlockInit
(
&
pWrapper
->
lock
,
NULL
);
snprintf
(
path
,
sizeof
(
path
),
"%s%s%s"
,
tsDataDir
,
TD_DIRSEP
,
pWrapper
->
name
);
pWrapper
->
path
=
strdup
(
path
);
...
...
@@ -223,7 +225,7 @@ SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
SMgmtWrapper
*
pWrapper
=
&
pDnode
->
wrappers
[
ntype
];
SMgmtWrapper
*
pRetWrapper
=
pWrapper
;
taos
RLockLatch
(
&
pWrapper
->
latch
);
taos
ThreadRwlockRdlock
(
&
pWrapper
->
lock
);
if
(
pWrapper
->
deployed
)
{
int32_t
refCount
=
atomic_add_fetch_32
(
&
pWrapper
->
refCount
,
1
);
dTrace
(
"node:%s, is acquired, ref:%d"
,
pWrapper
->
name
,
refCount
);
...
...
@@ -231,7 +233,7 @@ SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
terrno
=
TSDB_CODE_NODE_NOT_DEPLOYED
;
pRetWrapper
=
NULL
;
}
taos
RUnLockLatch
(
&
pWrapper
->
latch
);
taos
ThreadRwlockUnlock
(
&
pWrapper
->
lock
);
return
pRetWrapper
;
}
...
...
@@ -239,7 +241,7 @@ SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
int32_t
dmMarkWrapper
(
SMgmtWrapper
*
pWrapper
)
{
int32_t
code
=
0
;
taos
RLockLatch
(
&
pWrapper
->
latch
);
taos
ThreadRwlockRdlock
(
&
pWrapper
->
lock
);
if
(
pWrapper
->
deployed
||
(
InParentProc
(
pWrapper
)
&&
pWrapper
->
required
))
{
int32_t
refCount
=
atomic_add_fetch_32
(
&
pWrapper
->
refCount
,
1
);
dTrace
(
"node:%s, is marked, ref:%d"
,
pWrapper
->
name
,
refCount
);
...
...
@@ -247,7 +249,7 @@ int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {
terrno
=
TSDB_CODE_NODE_NOT_DEPLOYED
;
code
=
-
1
;
}
taos
RUnLockLatch
(
&
pWrapper
->
latch
);
taos
ThreadRwlockUnlock
(
&
pWrapper
->
lock
);
return
code
;
}
...
...
@@ -255,9 +257,9 @@ int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {
void
dmReleaseWrapper
(
SMgmtWrapper
*
pWrapper
)
{
if
(
pWrapper
==
NULL
)
return
;
taos
RLockLatch
(
&
pWrapper
->
latch
);
taos
ThreadRwlockRdlock
(
&
pWrapper
->
lock
);
int32_t
refCount
=
atomic_sub_fetch_32
(
&
pWrapper
->
refCount
,
1
);
taos
RUnLockLatch
(
&
pWrapper
->
latch
);
taos
ThreadRwlockUnlock
(
&
pWrapper
->
lock
);
dTrace
(
"node:%s, is released, ref:%d"
,
pWrapper
->
name
,
refCount
);
}
...
...
source/dnode/mgmt/node_mgmt/src/dmNodes.c
浏览文件 @
a9aa8e58
...
...
@@ -186,12 +186,12 @@ void dmCloseNode(SMgmtWrapper *pWrapper) {
}
}
taos
WLockLatch
(
&
pWrapper
->
latch
);
taos
ThreadRwlockWrlock
(
&
pWrapper
->
lock
);
if
(
pWrapper
->
pMgmt
!=
NULL
)
{
(
*
pWrapper
->
func
.
closeFp
)(
pWrapper
->
pMgmt
);
pWrapper
->
pMgmt
=
NULL
;
}
taos
WUnLockLatch
(
&
pWrapper
->
latch
);
taos
ThreadRwlockUnlock
(
&
pWrapper
->
lock
);
if
(
!
OnlyInSingleProc
(
pWrapper
))
{
dmCleanupProc
(
pWrapper
);
...
...
source/dnode/mgmt/node_util/inc/dmUtil.h
浏览文件 @
a9aa8e58
...
...
@@ -94,18 +94,18 @@ typedef void (*GetVnodeLoadsFp)();
typedef
void
(
*
GetMnodeLoadsFp
)();
typedef
struct
{
int32_t
dnodeId
;
int64_t
clusterId
;
int64_t
dnodeVer
;
int64_t
updateTime
;
int64_t
rebootTime
;
bool
dropped
;
bool
stopped
;
SEpSet
mnodeEps
;
SArray
*
dnodeEps
;
SHashObj
*
dnodeHash
;
SRWLatch
latch
;
SMsgCb
msgCb
;
int32_t
dnodeId
;
int64_t
clusterId
;
int64_t
dnodeVer
;
int64_t
updateTime
;
int64_t
rebootTime
;
bool
dropped
;
bool
stopped
;
SEpSet
mnodeEps
;
SArray
*
dnodeEps
;
SHashObj
*
dnodeHash
;
TdThreadRwlock
lock
;
SMsgCb
msgCb
;
}
SDnodeData
;
typedef
struct
{
...
...
source/dnode/mgmt/node_util/src/dmEps.c
浏览文件 @
a9aa8e58
...
...
@@ -21,7 +21,7 @@ static bool dmIsEpChanged(SDnodeData *pData, int32_t dnodeId, const char *ep);
static
void
dmResetEps
(
SDnodeData
*
pData
,
SArray
*
dnodeEps
);
static
void
dmGetDnodeEp
(
SDnodeData
*
pData
,
int32_t
dnodeId
,
char
*
pEp
,
char
*
pFqdn
,
uint16_t
*
pPort
)
{
taos
RLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockRdlock
(
&
pData
->
lock
);
SDnodeEp
*
pDnodeEp
=
taosHashGet
(
pData
->
dnodeHash
,
&
dnodeId
,
sizeof
(
int32_t
));
if
(
pDnodeEp
!=
NULL
)
{
...
...
@@ -36,7 +36,7 @@ static void dmGetDnodeEp(SDnodeData *pData, int32_t dnodeId, char *pEp, char *pF
}
}
taos
RUnLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockUnlock
(
&
pData
->
lock
);
}
int32_t
dmReadEps
(
SDnodeData
*
pData
)
{
...
...
@@ -232,7 +232,7 @@ void dmUpdateEps(SDnodeData *pData, SArray *eps) {
int32_t
numOfEps
=
taosArrayGetSize
(
eps
);
if
(
numOfEps
<=
0
)
return
;
taos
WLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockWrlock
(
&
pData
->
lock
);
int32_t
numOfEpsOld
=
(
int32_t
)
taosArrayGetSize
(
pData
->
dnodeEps
);
if
(
numOfEps
!=
numOfEpsOld
)
{
...
...
@@ -246,7 +246,7 @@ void dmUpdateEps(SDnodeData *pData, SArray *eps) {
}
}
taos
WUnLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockUnlock
(
&
pData
->
lock
);
}
static
void
dmResetEps
(
SDnodeData
*
pData
,
SArray
*
dnodeEps
)
{
...
...
@@ -292,7 +292,7 @@ static void dmPrintEps(SDnodeData *pData) {
static
bool
dmIsEpChanged
(
SDnodeData
*
pData
,
int32_t
dnodeId
,
const
char
*
ep
)
{
bool
changed
=
false
;
if
(
dnodeId
==
0
)
return
changed
;
taos
RLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockRdlock
(
&
pData
->
lock
);
SDnodeEp
*
pDnodeEp
=
taosHashGet
(
pData
->
dnodeHash
,
&
dnodeId
,
sizeof
(
int32_t
));
if
(
pDnodeEp
!=
NULL
)
{
...
...
@@ -304,24 +304,23 @@ static bool dmIsEpChanged(SDnodeData *pData, int32_t dnodeId, const char *ep) {
}
}
taos
RUnLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockUnlock
(
&
pData
->
lock
);
return
changed
;
}
void
dmGetMnodeEpSet
(
SDnodeData
*
pData
,
SEpSet
*
pEpSet
)
{
taos
RLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockRdlock
(
&
pData
->
lock
);
*
pEpSet
=
pData
->
mnodeEps
;
taos
RUnLockLatch
(
&
pData
->
latch
);
taos
ThreadRwlockUnlock
(
&
pData
->
lock
);
}
void
dmSetMnodeEpSet
(
SDnodeData
*
pData
,
SEpSet
*
pEpSet
)
{
dInfo
(
"mnode is changed, num:%d use:%d"
,
pEpSet
->
numOfEps
,
pEpSet
->
inUse
);
taosWLockLatch
(
&
pData
->
latch
);
taosThreadRwlockWrlock
(
&
pData
->
lock
);
pData
->
mnodeEps
=
*
pEpSet
;
taosThreadRwlockUnlock
(
&
pData
->
lock
);
dInfo
(
"mnode is changed, num:%d use:%d"
,
pEpSet
->
numOfEps
,
pEpSet
->
inUse
);
for
(
int32_t
i
=
0
;
i
<
pEpSet
->
numOfEps
;
++
i
)
{
dInfo
(
"mnode index:%d %s:%u"
,
i
,
pEpSet
->
eps
[
i
].
fqdn
,
pEpSet
->
eps
[
i
].
port
);
}
taosWUnLockLatch
(
&
pData
->
latch
);
}
source/dnode/mnode/sdb/src/sdb.c
浏览文件 @
a9aa8e58
...
...
@@ -48,7 +48,7 @@ SSdb *sdbInit(SSdbOpt *pOption) {
}
for
(
ESdbType
i
=
0
;
i
<
SDB_MAX
;
++
i
)
{
taos
InitRWLatch
(
&
pSdb
->
locks
[
i
]
);
taos
ThreadRwlockInit
(
&
pSdb
->
locks
[
i
],
NULL
);
pSdb
->
maxId
[
i
]
=
0
;
pSdb
->
tableVer
[
i
]
=
0
;
pSdb
->
keyTypes
[
i
]
=
SDB_KEY_INT32
;
...
...
@@ -98,7 +98,10 @@ void sdbCleanup(SSdb *pSdb) {
taosHashClear
(
hash
);
taosHashCleanup
(
hash
);
taosThreadRwlockDestroy
(
&
pSdb
->
locks
[
i
]);
pSdb
->
hashObjs
[
i
]
=
NULL
;
memset
(
&
pSdb
->
locks
[
i
],
0
,
sizeof
(
pSdb
->
locks
[
i
]));
mDebug
(
"sdb table:%s is cleaned up"
,
sdbTableName
(
i
));
}
...
...
@@ -134,7 +137,6 @@ int32_t sdbSetTable(SSdb *pSdb, SSdbTable table) {
pSdb
->
maxId
[
sdbType
]
=
0
;
pSdb
->
hashObjs
[
sdbType
]
=
hash
;
taosInitRWLatch
(
&
pSdb
->
locks
[
sdbType
]);
mDebug
(
"sdb table:%s is initialized"
,
sdbTableName
(
sdbType
));
return
0
;
...
...
source/dnode/mnode/sdb/src/sdbFile.c
浏览文件 @
a9aa8e58
...
...
@@ -257,8 +257,8 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
mTrace
(
"write %s to file, total %d rows"
,
sdbTableName
(
i
),
sdbGetSize
(
pSdb
,
i
));
SHashObj
*
hash
=
pSdb
->
hashObjs
[
i
];
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
i
];
taos
WLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
i
];
taos
ThreadRwlockWrlock
(
pLock
);
SSdbRow
**
ppRow
=
taosHashIterate
(
hash
,
NULL
);
while
(
ppRow
!=
NULL
)
{
...
...
@@ -303,7 +303,7 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
sdbFreeRaw
(
pRaw
);
ppRow
=
taosHashIterate
(
hash
,
ppRow
);
}
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
}
if
(
code
==
0
)
{
...
...
source/dnode/mnode/sdb/src/sdbHash.c
浏览文件 @
a9aa8e58
...
...
@@ -129,12 +129,12 @@ static int32_t sdbGetkeySize(SSdb *pSdb, ESdbType type, const void *pKey) {
}
static
int32_t
sdbInsertRow
(
SSdb
*
pSdb
,
SHashObj
*
hash
,
SSdbRaw
*
pRaw
,
SSdbRow
*
pRow
,
int32_t
keySize
)
{
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
WLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
ThreadRwlockWrlock
(
pLock
);
SSdbRow
*
pOldRow
=
taosHashGet
(
hash
,
pRow
->
pObj
,
keySize
);
if
(
pOldRow
!=
NULL
)
{
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
sdbFreeRow
(
pSdb
,
pRow
,
false
);
terrno
=
TSDB_CODE_SDB_OBJ_ALREADY_THERE
;
return
terrno
;
...
...
@@ -145,13 +145,13 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
sdbPrintOper
(
pSdb
,
pRow
,
"insert"
);
if
(
taosHashPut
(
hash
,
pRow
->
pObj
,
keySize
,
&
pRow
,
sizeof
(
void
*
))
!=
0
)
{
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
sdbFreeRow
(
pSdb
,
pRow
,
false
);
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
terrno
;
}
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
int32_t
code
=
0
;
SdbInsertFp
insertFp
=
pSdb
->
insertFps
[
pRow
->
type
];
...
...
@@ -159,9 +159,9 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
code
=
(
*
insertFp
)(
pSdb
,
pRow
->
pObj
);
if
(
code
!=
0
)
{
code
=
terrno
;
taos
WLockLatch
(
pLock
);
taos
ThreadRwlockWrlock
(
pLock
);
taosHashRemove
(
hash
,
pRow
->
pObj
,
keySize
);
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
sdbFreeRow
(
pSdb
,
pRow
,
false
);
terrno
=
code
;
return
terrno
;
...
...
@@ -180,19 +180,19 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
}
static
int32_t
sdbUpdateRow
(
SSdb
*
pSdb
,
SHashObj
*
hash
,
SSdbRaw
*
pRaw
,
SSdbRow
*
pNewRow
,
int32_t
keySize
)
{
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
pNewRow
->
type
];
taos
WLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
pNewRow
->
type
];
taos
ThreadRwlockWrlock
(
pLock
);
SSdbRow
**
ppOldRow
=
taosHashGet
(
hash
,
pNewRow
->
pObj
,
keySize
);
if
(
ppOldRow
==
NULL
||
*
ppOldRow
==
NULL
)
{
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
return
sdbInsertRow
(
pSdb
,
hash
,
pRaw
,
pNewRow
,
keySize
);
}
SSdbRow
*
pOldRow
=
*
ppOldRow
;
pOldRow
->
status
=
pRaw
->
status
;
sdbPrintOper
(
pSdb
,
pOldRow
,
"update"
);
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
int32_t
code
=
0
;
SdbUpdateFp
updateFp
=
pSdb
->
updateFps
[
pNewRow
->
type
];
...
...
@@ -207,12 +207,12 @@ static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
}
static
int32_t
sdbDeleteRow
(
SSdb
*
pSdb
,
SHashObj
*
hash
,
SSdbRaw
*
pRaw
,
SSdbRow
*
pRow
,
int32_t
keySize
)
{
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
WLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
ThreadRwlockWrlock
(
pLock
);
SSdbRow
**
ppOldRow
=
taosHashGet
(
hash
,
pRow
->
pObj
,
keySize
);
if
(
ppOldRow
==
NULL
||
*
ppOldRow
==
NULL
)
{
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
sdbFreeRow
(
pSdb
,
pRow
,
false
);
terrno
=
TSDB_CODE_SDB_OBJ_NOT_THERE
;
return
terrno
;
...
...
@@ -223,7 +223,7 @@ static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
sdbPrintOper
(
pSdb
,
pOldRow
,
"delete"
);
taosHashRemove
(
hash
,
pOldRow
->
pObj
,
keySize
);
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
pSdb
->
tableVer
[
pOldRow
->
type
]
++
;
sdbFreeRow
(
pSdb
,
pRow
,
false
);
...
...
@@ -278,12 +278,12 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
void
*
pRet
=
NULL
;
int32_t
keySize
=
sdbGetkeySize
(
pSdb
,
type
,
pKey
);
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
RLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
ThreadRwlockRdlock
(
pLock
);
SSdbRow
**
ppRow
=
taosHashGet
(
hash
,
pKey
,
keySize
);
if
(
ppRow
==
NULL
||
*
ppRow
==
NULL
)
{
taos
RUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
terrno
=
TSDB_CODE_SDB_OBJ_NOT_THERE
;
return
NULL
;
}
...
...
@@ -306,13 +306,13 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
break
;
}
taos
RUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
return
pRet
;
}
static
void
sdbCheckRow
(
SSdb
*
pSdb
,
SSdbRow
*
pRow
)
{
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
WLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
ThreadRwlockWrlock
(
pLock
);
int32_t
ref
=
atomic_load_32
(
&
pRow
->
refCount
);
sdbPrintOper
(
pSdb
,
pRow
,
"check"
);
...
...
@@ -320,7 +320,7 @@ static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow) {
sdbFreeRow
(
pSdb
,
pRow
,
true
);
}
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
}
void
sdbRelease
(
SSdb
*
pSdb
,
void
*
pObj
)
{
...
...
@@ -329,8 +329,8 @@ void sdbRelease(SSdb *pSdb, void *pObj) {
SSdbRow
*
pRow
=
(
SSdbRow
*
)((
char
*
)
pObj
-
sizeof
(
SSdbRow
));
if
(
pRow
->
type
>=
SDB_MAX
)
return
;
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
WLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
ThreadRwlockWrlock
(
pLock
);
int32_t
ref
=
atomic_sub_fetch_32
(
&
pRow
->
refCount
,
1
);
sdbPrintOper
(
pSdb
,
pRow
,
"release"
);
...
...
@@ -338,7 +338,7 @@ void sdbRelease(SSdb *pSdb, void *pObj) {
sdbFreeRow
(
pSdb
,
pRow
,
true
);
}
taos
WUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
}
void
*
sdbFetch
(
SSdb
*
pSdb
,
ESdbType
type
,
void
*
pIter
,
void
**
ppObj
)
{
...
...
@@ -347,8 +347,8 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
SHashObj
*
hash
=
sdbGetHash
(
pSdb
,
type
);
if
(
hash
==
NULL
)
return
NULL
;
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
RLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
ThreadRwlockRdlock
(
pLock
);
SSdbRow
**
ppRow
=
taosHashIterate
(
hash
,
pIter
);
while
(
ppRow
!=
NULL
)
{
...
...
@@ -363,7 +363,7 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
*
ppObj
=
pRow
->
pObj
;
break
;
}
taos
RUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
return
ppRow
;
}
...
...
@@ -374,18 +374,18 @@ void sdbCancelFetch(SSdb *pSdb, void *pIter) {
SHashObj
*
hash
=
sdbGetHash
(
pSdb
,
pRow
->
type
);
if
(
hash
==
NULL
)
return
;
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
RLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
pRow
->
type
];
taos
ThreadRwlockRdlock
(
pLock
);
taosHashCancelIterate
(
hash
,
pIter
);
taos
RUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
}
void
sdbTraverse
(
SSdb
*
pSdb
,
ESdbType
type
,
sdbTraverseFp
fp
,
void
*
p1
,
void
*
p2
,
void
*
p3
)
{
SHashObj
*
hash
=
sdbGetHash
(
pSdb
,
type
);
if
(
hash
==
NULL
)
return
;
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
RLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
ThreadRwlockRdlock
(
pLock
);
SSdbRow
**
ppRow
=
taosHashIterate
(
hash
,
NULL
);
while
(
ppRow
!=
NULL
)
{
...
...
@@ -401,17 +401,17 @@ void sdbTraverse(SSdb *pSdb, ESdbType type, sdbTraverseFp fp, void *p1, void *p2
ppRow
=
taosHashIterate
(
hash
,
ppRow
);
}
taos
RUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
}
int32_t
sdbGetSize
(
SSdb
*
pSdb
,
ESdbType
type
)
{
SHashObj
*
hash
=
sdbGetHash
(
pSdb
,
type
);
if
(
hash
==
NULL
)
return
0
;
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
RLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
ThreadRwlockRdlock
(
pLock
);
int32_t
size
=
taosHashGetSize
(
hash
);
taos
RUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
return
size
;
}
...
...
@@ -424,8 +424,8 @@ int32_t sdbGetMaxId(SSdb *pSdb, ESdbType type) {
int32_t
maxId
=
0
;
SRWLatch
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
RLockLatch
(
pLock
);
TdThreadRwlock
*
pLock
=
&
pSdb
->
locks
[
type
];
taos
ThreadRwlockRdlock
(
pLock
);
SSdbRow
**
ppRow
=
taosHashIterate
(
hash
,
NULL
);
while
(
ppRow
!=
NULL
)
{
...
...
@@ -435,7 +435,7 @@ int32_t sdbGetMaxId(SSdb *pSdb, ESdbType type) {
ppRow
=
taosHashIterate
(
hash
,
ppRow
);
}
taos
RUnLockLatch
(
pLock
);
taos
ThreadRwlockUnlock
(
pLock
);
maxId
=
TMAX
(
maxId
,
pSdb
->
maxId
[
type
]);
return
maxId
+
1
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录