Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
慢慢CG
TDengine
提交
e4956e3f
T
TDengine
项目概览
慢慢CG
/
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看板
提交
e4956e3f
编写于
4月 12, 2020
作者:
J
Jeff Tao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
remove invalid read in hash.c
remove goto in tsched.c
上级
a9be7a85
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
55 addition
and
52 deletion
+55
-52
src/util/src/hash.c
src/util/src/hash.c
+1
-1
src/util/src/tsched.c
src/util/src/tsched.c
+54
-51
未找到文件。
src/util/src/hash.c
浏览文件 @
e4956e3f
...
...
@@ -298,7 +298,7 @@ SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool threadsafe) {
*/
static
SHashNode
*
doCreateHashNode
(
const
char
*
key
,
size_t
keyLen
,
const
char
*
pData
,
size_t
dataSize
,
uint32_t
hashVal
)
{
size_t
totalSize
=
dataSize
+
sizeof
(
SHashNode
)
+
keyLen
;
size_t
totalSize
=
dataSize
+
sizeof
(
SHashNode
)
+
keyLen
+
1
;
// one extra byte for null
SHashNode
*
pNewNode
=
calloc
(
1
,
totalSize
);
if
(
pNewNode
==
NULL
)
{
...
...
src/util/src/tsched.c
浏览文件 @
e4956e3f
...
...
@@ -40,68 +40,68 @@ static void *taosProcessSchedQueue(void *param);
static
void
taosDumpSchedulerStatus
(
void
*
qhandle
,
void
*
tmrId
);
void
*
taosInitScheduler
(
int
queueSize
,
int
numOfThreads
,
const
char
*
label
)
{
pthread_attr_t
attr
;
SSchedQueue
*
pSched
=
(
SSchedQueue
*
)
malloc
(
sizeof
(
SSchedQueue
));
SSchedQueue
*
pSched
=
(
SSchedQueue
*
)
calloc
(
sizeof
(
SSchedQueue
),
1
);
if
(
pSched
==
NULL
)
{
pError
(
"%s: no enough memory for pSched, reason: %s"
,
label
,
strerror
(
errno
));
goto
_error
;
pError
(
"%s: no enough memory for pSched"
,
label
);
return
NULL
;
}
pSched
->
queue
=
(
SSchedMsg
*
)
calloc
(
sizeof
(
SSchedMsg
),
queueSize
);
if
(
pSched
->
queue
==
NULL
)
{
pError
(
"%s: no enough memory for queue"
,
label
);
taosCleanUpScheduler
(
pSched
);
return
NULL
;
}
pSched
->
qthread
=
calloc
(
sizeof
(
pthread_t
),
numOfThreads
);
if
(
pSched
->
qthread
==
NULL
)
{
pError
(
"%s: no enough memory for qthread"
,
label
);
taosCleanUpScheduler
(
pSched
);
return
NULL
;
}
memset
(
pSched
,
0
,
sizeof
(
SSchedQueue
));
pSched
->
queueSize
=
queueSize
;
strncpy
(
pSched
->
label
,
label
,
sizeof
(
pSched
->
label
));
// fix buffer overflow
pSched
->
label
[
sizeof
(
pSched
->
label
)
-
1
]
=
'\0'
;
pSched
->
fullSlot
=
0
;
pSched
->
emptySlot
=
0
;
if
(
pthread_mutex_init
(
&
pSched
->
queueMutex
,
NULL
)
<
0
)
{
pError
(
"init %s:queueMutex failed, reason:%s"
,
pSched
->
label
,
strerror
(
errno
));
goto
_error
;
pError
(
"init %s:queueMutex failed(%s)"
,
label
,
strerror
(
errno
));
taosCleanUpScheduler
(
pSched
);
return
NULL
;
}
if
(
tsem_init
(
&
pSched
->
emptySem
,
0
,
(
unsigned
int
)
pSched
->
queueSize
)
!=
0
)
{
pError
(
"init %s:empty semaphore failed, reason:%s"
,
pSched
->
label
,
strerror
(
errno
));
goto
_error
;
pError
(
"init %s:empty semaphore failed(%s)"
,
label
,
strerror
(
errno
));
taosCleanUpScheduler
(
pSched
);
return
NULL
;
}
if
(
tsem_init
(
&
pSched
->
fullSem
,
0
,
0
)
!=
0
)
{
pError
(
"init %s:full semaphore failed, reason:%s"
,
pSched
->
label
,
strerror
(
errno
));
goto
_error
;
}
if
((
pSched
->
queue
=
(
SSchedMsg
*
)
malloc
((
size_t
)
pSched
->
queueSize
*
sizeof
(
SSchedMsg
)))
==
NULL
)
{
pError
(
"%s: no enough memory for queue, reason:%s"
,
pSched
->
label
,
strerror
(
errno
));
goto
_error
;
pError
(
"init %s:full semaphore failed(%s)"
,
label
,
strerror
(
errno
));
taosCleanUpScheduler
(
pSched
);
return
NULL
;
}
memset
(
pSched
->
queue
,
0
,
(
size_t
)
pSched
->
queueSize
*
sizeof
(
SSchedMsg
));
pSched
->
fullSlot
=
0
;
pSched
->
emptySlot
=
0
;
pSched
->
qthread
=
malloc
(
sizeof
(
pthread_t
)
*
(
size_t
)
numOfThreads
);
if
(
pSched
->
qthread
==
NULL
)
{
pError
(
"%s: no enough memory for qthread, reason: %s"
,
pSched
->
label
,
strerror
(
errno
));
goto
_error
;
}
pthread_attr_init
(
&
attr
);
pthread_attr_setdetachstate
(
&
attr
,
PTHREAD_CREATE_JOINABLE
);
for
(
int
i
=
0
;
i
<
numOfThreads
;
++
i
)
{
if
(
pthread_create
(
pSched
->
qthread
+
i
,
&
attr
,
taosProcessSchedQueue
,
(
void
*
)
pSched
)
!=
0
)
{
pError
(
"%s: failed to create rpc thread, reason:%s"
,
pSched
->
label
,
strerror
(
errno
));
goto
_error
;
pthread_attr_t
attr
;
pthread_attr_init
(
&
attr
);
pthread_attr_setdetachstate
(
&
attr
,
PTHREAD_CREATE_JOINABLE
);
int
code
=
pthread_create
(
pSched
->
qthread
+
i
,
&
attr
,
taosProcessSchedQueue
,
(
void
*
)
pSched
);
pthread_attr_destroy
(
&
attr
);
if
(
code
!=
0
)
{
pError
(
"%s: failed to create rpc thread(%s)"
,
label
,
strerror
(
errno
));
taosCleanUpScheduler
(
pSched
);
return
NULL
;
}
++
pSched
->
numOfThreads
;
}
pTrace
(
"%s scheduler is initialized, numOfThreads:%d"
,
pSched
->
label
,
pSched
->
numOfThreads
);
pthread_attr_destroy
(
&
attr
);
pTrace
(
"%s scheduler is initialized, numOfThreads:%d"
,
label
,
pSched
->
numOfThreads
);
return
(
void
*
)
pSched
;
_error:
pthread_attr_destroy
(
&
attr
);
taosCleanUpScheduler
(
pSched
);
return
NULL
;
}
void
*
taosInitSchedulerWithInfo
(
int
queueSize
,
int
numOfThreads
,
const
char
*
label
,
void
*
tmrCtrl
)
{
...
...
@@ -126,21 +126,21 @@ void *taosProcessSchedQueue(void *param) {
pTrace
(
"wait %s fullSem was interrupted"
,
pSched
->
label
);
continue
;
}
pError
(
"wait %s fullSem failed
, errno:%d, reason:%s"
,
pSched
->
label
,
errno
,
strerror
(
errno
));
pError
(
"wait %s fullSem failed
(%s)"
,
pSched
->
label
,
strerror
(
errno
));
}
if
(
pthread_mutex_lock
(
&
pSched
->
queueMutex
)
!=
0
)
pError
(
"lock %s queueMutex failed
, reason:%s
"
,
pSched
->
label
,
strerror
(
errno
));
pError
(
"lock %s queueMutex failed
(%s)
"
,
pSched
->
label
,
strerror
(
errno
));
msg
=
pSched
->
queue
[
pSched
->
fullSlot
];
memset
(
pSched
->
queue
+
pSched
->
fullSlot
,
0
,
sizeof
(
SSchedMsg
));
pSched
->
fullSlot
=
(
pSched
->
fullSlot
+
1
)
%
pSched
->
queueSize
;
if
(
pthread_mutex_unlock
(
&
pSched
->
queueMutex
)
!=
0
)
pError
(
"unlock %s queueMutex failed
, reason:%s
\n
"
,
pSched
->
label
,
strerror
(
errno
));
pError
(
"unlock %s queueMutex failed
(%s)
"
,
pSched
->
label
,
strerror
(
errno
));
if
(
tsem_post
(
&
pSched
->
emptySem
)
!=
0
)
pError
(
"post %s emptySem failed
, reason:%s
\n
"
,
pSched
->
label
,
strerror
(
errno
));
pError
(
"post %s emptySem failed
(%s)
"
,
pSched
->
label
,
strerror
(
errno
));
if
(
msg
.
fp
)
(
*
(
msg
.
fp
))(
&
msg
);
...
...
@@ -160,22 +160,23 @@ int taosScheduleTask(void *qhandle, SSchedMsg *pMsg) {
while
(
tsem_wait
(
&
pSched
->
emptySem
)
!=
0
)
{
if
(
errno
!=
EINTR
)
{
pError
(
"wait %s emptySem failed
, reason:%s
"
,
pSched
->
label
,
strerror
(
errno
));
pError
(
"wait %s emptySem failed
(%s)
"
,
pSched
->
label
,
strerror
(
errno
));
break
;
}
pTrace
(
"wait %s emptySem was interrupted"
,
pSched
->
label
);
}
if
(
pthread_mutex_lock
(
&
pSched
->
queueMutex
)
!=
0
)
pError
(
"lock %s queueMutex failed
, reason:%s
"
,
pSched
->
label
,
strerror
(
errno
));
pError
(
"lock %s queueMutex failed
(%s)
"
,
pSched
->
label
,
strerror
(
errno
));
pSched
->
queue
[
pSched
->
emptySlot
]
=
*
pMsg
;
pSched
->
emptySlot
=
(
pSched
->
emptySlot
+
1
)
%
pSched
->
queueSize
;
if
(
pthread_mutex_unlock
(
&
pSched
->
queueMutex
)
!=
0
)
pError
(
"unlock %s queueMutex failed
, reason:%s
"
,
pSched
->
label
,
strerror
(
errno
));
pError
(
"unlock %s queueMutex failed
(%s)
"
,
pSched
->
label
,
strerror
(
errno
));
if
(
tsem_post
(
&
pSched
->
fullSem
)
!=
0
)
pError
(
"post %s fullSem failed, reason:%s"
,
pSched
->
label
,
strerror
(
errno
));
if
(
tsem_post
(
&
pSched
->
fullSem
)
!=
0
)
pError
(
"post %s fullSem failed(%s)"
,
pSched
->
label
,
strerror
(
errno
));
return
0
;
}
...
...
@@ -185,10 +186,12 @@ void taosCleanUpScheduler(void *param) {
if
(
pSched
==
NULL
)
return
;
for
(
int
i
=
0
;
i
<
pSched
->
numOfThreads
;
++
i
)
{
pthread_cancel
(
pSched
->
qthread
[
i
]);
if
(
pSched
->
qthread
[
i
])
pthread_cancel
(
pSched
->
qthread
[
i
]);
}
for
(
int
i
=
0
;
i
<
pSched
->
numOfThreads
;
++
i
)
{
pthread_join
(
pSched
->
qthread
[
i
],
NULL
);
if
(
pSched
->
qthread
[
i
])
pthread_join
(
pSched
->
qthread
[
i
],
NULL
);
}
tsem_destroy
(
&
pSched
->
emptySem
);
...
...
@@ -199,8 +202,8 @@ void taosCleanUpScheduler(void *param) {
taosTmrStopA
(
&
pSched
->
pTimer
);
}
free
(
pSched
->
queue
);
free
(
pSched
->
qthread
);
if
(
pSched
->
queue
)
free
(
pSched
->
queue
);
if
(
pSched
->
qthread
)
free
(
pSched
->
qthread
);
free
(
pSched
);
// fix memory leak
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录