Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
2dot5
ClickHouse
提交
b1486015
C
ClickHouse
项目概览
2dot5
/
ClickHouse
通知
3
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
C
ClickHouse
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
b1486015
编写于
8月 03, 2019
作者:
A
alexey-milovidov
提交者:
GitHub
8月 03, 2019
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #6301 from yandex/fix-thread-pool-hang-if-full
Fixed the possibility of hanging queries when server is overloaded
上级
f7fbdab1
9f6e26f1
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
128 addition
and
5 deletion
+128
-5
dbms/src/Common/ThreadPool.cpp
dbms/src/Common/ThreadPool.cpp
+32
-2
dbms/src/Common/ThreadPool.h
dbms/src/Common/ThreadPool.h
+7
-3
dbms/src/Common/tests/gtest_thread_pool_global_full.cpp
dbms/src/Common/tests/gtest_thread_pool_global_full.cpp
+89
-0
未找到文件。
dbms/src/Common/ThreadPool.cpp
浏览文件 @
b1486015
#include <Common/ThreadPool.h>
#include <Common/Exception.h>
#include <iostream>
#include <type_traits>
...
...
@@ -34,6 +33,28 @@ ThreadPoolImpl<Thread>::ThreadPoolImpl(size_t max_threads, size_t max_free_threa
{
}
template
<
typename
Thread
>
void
ThreadPoolImpl
<
Thread
>::
setMaxThreads
(
size_t
value
)
{
std
::
lock_guard
lock
(
mutex
);
max_threads
=
value
;
}
template
<
typename
Thread
>
void
ThreadPoolImpl
<
Thread
>::
setMaxFreeThreads
(
size_t
value
)
{
std
::
lock_guard
lock
(
mutex
);
max_free_threads
=
value
;
}
template
<
typename
Thread
>
void
ThreadPoolImpl
<
Thread
>::
setQueueSize
(
size_t
value
)
{
std
::
lock_guard
lock
(
mutex
);
queue_size
=
value
;
}
template
<
typename
Thread
>
template
<
typename
ReturnType
>
ReturnType
ThreadPoolImpl
<
Thread
>::
scheduleImpl
(
Job
job
,
int
priority
,
std
::
optional
<
uint64_t
>
wait_microseconds
)
...
...
@@ -59,7 +80,7 @@ ReturnType ThreadPoolImpl<Thread>::scheduleImpl(Job job, int priority, std::opti
auto
pred
=
[
this
]
{
return
!
queue_size
||
scheduled_jobs
<
queue_size
||
shutdown
;
};
if
(
wait_microseconds
)
if
(
wait_microseconds
)
/// Check for optional. Condition is true if the optional is set and the value is zero.
{
if
(
!
job_finished
.
wait_for
(
lock
,
std
::
chrono
::
microseconds
(
*
wait_microseconds
),
pred
))
return
on_error
();
...
...
@@ -83,6 +104,15 @@ ReturnType ThreadPoolImpl<Thread>::scheduleImpl(Job job, int priority, std::opti
catch
(...)
{
threads
.
pop_front
();
/// Remove the job and return error to caller.
/// Note that if we have allocated at least one thread, we may continue
/// (one thread is enough to process all jobs).
/// But this condition indicate an error nevertheless and better to refuse.
jobs
.
pop
();
--
scheduled_jobs
;
return
on_error
();
}
}
}
...
...
dbms/src/Common/ThreadPool.h
浏览文件 @
b1486015
...
...
@@ -60,14 +60,18 @@ public:
/// Returns number of running and scheduled jobs.
size_t
active
()
const
;
void
setMaxThreads
(
size_t
value
);
void
setMaxFreeThreads
(
size_t
value
);
void
setQueueSize
(
size_t
value
);
private:
mutable
std
::
mutex
mutex
;
std
::
condition_variable
job_finished
;
std
::
condition_variable
new_job_or_shutdown
;
const
size_t
max_threads
;
const
size_t
max_free_threads
;
const
size_t
queue_size
;
size_t
max_threads
;
size_t
max_free_threads
;
size_t
queue_size
;
size_t
scheduled_jobs
=
0
;
bool
shutdown
=
false
;
...
...
dbms/src/Common/tests/gtest_thread_pool_global_full.cpp
0 → 100644
浏览文件 @
b1486015
#include <atomic>
#include <Common/ThreadPool.h>
#include <gtest/gtest.h>
/// Test what happens if local ThreadPool cannot create a ThreadFromGlobalPool.
/// There was a bug: if local ThreadPool cannot allocate even a single thread,
/// the job will be scheduled but never get executed.
TEST
(
ThreadPool
,
GlobalFull1
)
{
GlobalThreadPool
&
global_pool
=
GlobalThreadPool
::
instance
();
static
constexpr
size_t
capacity
=
5
;
global_pool
.
setMaxThreads
(
capacity
);
global_pool
.
setMaxFreeThreads
(
1
);
global_pool
.
setQueueSize
(
capacity
);
global_pool
.
wait
();
std
::
atomic
<
size_t
>
counter
=
0
;
static
constexpr
size_t
num_jobs
=
capacity
+
1
;
auto
func
=
[
&
]
{
++
counter
;
while
(
counter
!=
num_jobs
)
{}
};
ThreadPool
pool
(
num_jobs
);
for
(
size_t
i
=
0
;
i
<
capacity
;
++
i
)
pool
.
schedule
(
func
);
for
(
size_t
i
=
capacity
;
i
<
num_jobs
;
++
i
)
{
EXPECT_THROW
(
pool
.
schedule
(
func
),
DB
::
Exception
);
++
counter
;
}
pool
.
wait
();
EXPECT_EQ
(
counter
,
num_jobs
);
global_pool
.
setMaxThreads
(
10000
);
global_pool
.
setMaxFreeThreads
(
1000
);
global_pool
.
setQueueSize
(
10000
);
}
TEST
(
ThreadPool
,
GlobalFull2
)
{
GlobalThreadPool
&
global_pool
=
GlobalThreadPool
::
instance
();
static
constexpr
size_t
capacity
=
5
;
global_pool
.
setMaxThreads
(
capacity
);
global_pool
.
setMaxFreeThreads
(
1
);
global_pool
.
setQueueSize
(
capacity
);
/// ThreadFromGlobalPool from local thread pools from previous test case have exited
/// but their threads from global_pool may not have finished (they still have to exit).
/// If we will not wait here, we can get "Cannot schedule a task exception" earlier than we expect in this test.
global_pool
.
wait
();
std
::
atomic
<
size_t
>
counter
=
0
;
auto
func
=
[
&
]
{
++
counter
;
while
(
counter
!=
capacity
+
1
)
{}
};
ThreadPool
pool
(
capacity
,
0
,
capacity
);
for
(
size_t
i
=
0
;
i
<
capacity
;
++
i
)
pool
.
schedule
(
func
);
ThreadPool
another_pool
(
1
);
EXPECT_THROW
(
another_pool
.
schedule
(
func
),
DB
::
Exception
);
++
counter
;
pool
.
wait
();
global_pool
.
wait
();
for
(
size_t
i
=
0
;
i
<
capacity
;
++
i
)
another_pool
.
schedule
([
&
]
{
++
counter
;
});
another_pool
.
wait
();
EXPECT_EQ
(
counter
,
capacity
*
2
+
1
);
global_pool
.
setMaxThreads
(
10000
);
global_pool
.
setMaxFreeThreads
(
1000
);
global_pool
.
setQueueSize
(
10000
);
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录