thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-30 05:30:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

remove unnessary thread

Browse Source
This commit is contained in:
Vxider 2019-12-19 11:26:23 +08:00
parent 0490510b0b
commit 3725b6cca3
1 changed files with 18 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
dbms/src/DataStreams/copyData.cpp
View File

@ -55,7 +55,7 @@ void copyDataImpl(IBlockInputStream & from, IBlockOutputStream & to, TCancelCall
inline void doNothing(const Block &) {} inline void doNothing(const Block &) {}
void copyDataImpl(BlockInputStreams & froms, BlockOutputStreams & tos) static void copyDataImpl(BlockInputStreams & froms, BlockOutputStreams & tos)
{ {
if (froms.size() == tos.size()) if (froms.size() == tos.size())
{ {
@ -79,25 +79,6 @@ void copyDataImpl(BlockInputStreams & froms, BlockOutputStreams & tos)
else else
{ {
ConcurrentBoundedQueue<Block> queue(froms.size()); ConcurrentBoundedQueue<Block> queue(froms.size());
ThreadFromGlobalPool from_threads([&]()
{
std::vector<ThreadFromGlobalPool> from_threads_;
from_threads_.reserve(froms.size());
for (auto & from : froms)
{
from_threads_.emplace_back([&queue, from]()
{
from->readPrefix();
while (Block block = from->read())
queue.push(block);
from->readSuffix();
});
}
for (auto & thread : from_threads_)
thread.join();
for (size_t i = 0; i < tos.size(); i++)
queue.push({});
});
std::vector<ThreadFromGlobalPool> to_threads; std::vector<ThreadFromGlobalPool> to_threads;
for (auto & to : tos) for (auto & to : tos)
{ {
@ -116,7 +97,23 @@ void copyDataImpl(BlockInputStreams & froms, BlockOutputStreams & tos)
}); });
} }
from_threads.join(); std::vector<ThreadFromGlobalPool> from_threads_;
from_threads_.reserve(froms.size());
for (auto & from : froms)
{
from_threads_.emplace_back([&queue, from]()
{
from->readPrefix();
while (Block block = from->read())
queue.push(block);
from->readSuffix();
});
}
for (auto & thread : from_threads_)
thread.join();
for (size_t i = 0; i < tos.size(); i++)
queue.push({});
for (auto & thread : to_threads) for (auto & thread : to_threads)
thread.join(); thread.join();
} }