thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-21 23:21:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

dbms: more uniform distribution of work among threads [#METR-2944].

Browse Source
This commit is contained in:
Alexey Milovidov 2015-03-08 01:00:58 +03:00
parent 5d19f28d05
commit ae4c458c5b
5 changed files with 22 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
dbms/include/DB/DataStreams/ParallelInputsProcessor.h
View File

@ -1,7 +1,7 @@
#pragma once #pragma once
#include <list> #include <list>
#include <queue> #include <stack>
#include <atomic> #include <atomic>
#include <thread> #include <thread>
#include <mutex> #include <mutex>
@ -49,7 +49,7 @@ public:
: inputs(inputs_), max_threads(std::min(inputs_.size(), max_threads_)), handler(handler_) : inputs(inputs_), max_threads(std::min(inputs_.size(), max_threads_)), handler(handler_)
{ {
for (size_t i = 0; i < inputs_.size(); ++i) for (size_t i = 0; i < inputs_.size(); ++i)
input_queue.emplace(inputs_[i], i); input_stack.emplace(inputs_[i], i);
} }
~ParallelInputsProcessor() ~ParallelInputsProcessor()
@ -162,16 +162,16 @@ private:
/// Выбираем следующий источник. /// Выбираем следующий источник.
{ {
std::lock_guard<std::mutex> lock(input_queue_mutex); std::lock_guard<std::mutex> lock(input_stack_mutex);
/// Если свободных источников нет, то этот поток больше не нужен. (Но другие потоки могут работать со своими источниками.) /// Если свободных источников нет, то этот поток больше не нужен. (Но другие потоки могут работать со своими источниками.)
if (input_queue.empty()) if (input_stack.empty())
break; break;
input = input_queue.front(); input = input_stack.top();
/// Убираем источник из очереди доступных источников. /// Убираем источник из очереди доступных источников.
input_queue.pop(); input_stack.pop();
} }
/// Основная работа. /// Основная работа.
@ -183,15 +183,15 @@ private:
/// Если этот источник ещё не иссяк, то положим полученный блок в очередь готовых. /// Если этот источник ещё не иссяк, то положим полученный блок в очередь готовых.
{ {
std::lock_guard<std::mutex> lock(input_queue_mutex); std::lock_guard<std::mutex> lock(input_stack_mutex);
if (block) if (block)
{ {
input_queue.push(input); input_stack.push(input);
} }
else else
{ {
if (input_queue.empty()) if (input_stack.empty())
break; break;
} }
} }
@ -214,12 +214,15 @@ private:
typedef std::vector<std::thread> ThreadsData; typedef std::vector<std::thread> ThreadsData;
ThreadsData threads; ThreadsData threads;
/// Очередь доступных источников, которые не заняты каким-либо потоком в данный момент. /** Стек доступных источников, которые не заняты каким-либо потоком в данный момент.
typedef std::queue<InputData> InputQueue; * Стек вместо очереди - чтобы выполнять работу по чтению одного источника более последовательно.
InputQueue input_queue; * То есть, продолжать обработку источника, который недавно обрабатывался.
*/
typedef std::stack<InputData> InputStack;
InputStack input_stack;
/// Для операций с input_queue. /// Для операций с input_stack.
std::mutex input_queue_mutex; std::mutex input_stack_mutex;
/// Сколько источников иссякло. /// Сколько источников иссякло.
std::atomic<size_t> active_threads { 0 }; std::atomic<size_t> active_threads { 0 };

7
dbms/src/Interpreters/InterpreterSelectQuery.cpp
View File

@ -12,7 +12,6 @@
#include <DB/DataStreams/DistinctBlockInputStream.h> #include <DB/DataStreams/DistinctBlockInputStream.h>
#include <DB/DataStreams/NullBlockInputStream.h> #include <DB/DataStreams/NullBlockInputStream.h>
#include <DB/DataStreams/TotalsHavingBlockInputStream.h> #include <DB/DataStreams/TotalsHavingBlockInputStream.h>
#include <DB/DataStreams/narrowBlockInputStreams.h>
#include <DB/DataStreams/copyData.h> #include <DB/DataStreams/copyData.h>
#include <DB/DataStreams/CreatingSetsBlockInputStream.h> #include <DB/DataStreams/CreatingSetsBlockInputStream.h>
#include <DB/DataStreams/MaterializingBlockInputStream.h> #include <DB/DataStreams/MaterializingBlockInputStream.h>
@ -697,12 +696,6 @@ QueryProcessingStage::Enum InterpreterSelectQuery::executeFetchColumns(BlockInpu
streams.push_back(interpreter_subquery->execute()); streams.push_back(interpreter_subquery->execute());
} }
/** Если истчоников слишком много, то склеим их в max_threads источников.
* (Иначе действия в каждом маленьком источнике, а затем объединение состояний, слишком неэффективно.)
*/
if (streams.size() > settings.max_threads)
streams = narrowBlockInputStreams(streams, settings.max_threads);
/** Установка ограничений и квоты на чтение данных, скорость и время выполнения запроса. /** Установка ограничений и квоты на чтение данных, скорость и время выполнения запроса.
* Такие ограничения проверяются на сервере-инициаторе запроса, а не на удалённых серверах. * Такие ограничения проверяются на сервере-инициаторе запроса, а не на удалённых серверах.
* Потому что сервер-инициатор имеет суммарные данные о выполнении запроса на всех серверах. * Потому что сервер-инициатор имеет суммарные данные о выполнении запроса на всех серверах.

17
dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
View File

@ -4,7 +4,6 @@
#include <DB/Parsers/ASTIdentifier.h> #include <DB/Parsers/ASTIdentifier.h>
#include <DB/DataStreams/ExpressionBlockInputStream.h> #include <DB/DataStreams/ExpressionBlockInputStream.h>
#include <DB/DataStreams/FilterBlockInputStream.h> #include <DB/DataStreams/FilterBlockInputStream.h>
#include <DB/DataStreams/ConcatBlockInputStream.h>
#include <DB/DataStreams/CollapsingFinalBlockInputStream.h> #include <DB/DataStreams/CollapsingFinalBlockInputStream.h>
#include <DB/DataStreams/AddingConstColumnBlockInputStream.h> #include <DB/DataStreams/AddingConstColumnBlockInputStream.h>
#include <DB/DataStreams/CreatingSetsBlockInputStream.h> #include <DB/DataStreams/CreatingSetsBlockInputStream.h>
@ -374,7 +373,6 @@ BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreads(
for (size_t i = 0; i < threads && !parts.empty(); ++i) for (size_t i = 0; i < threads && !parts.empty(); ++i)
{ {
size_t need_marks = min_marks_per_thread; size_t need_marks = min_marks_per_thread;
BlockInputStreams streams;
/// Цикл по кускам. /// Цикл по кускам.
while (need_marks > 0 && !parts.empty()) while (need_marks > 0 && !parts.empty())
@ -427,7 +425,7 @@ BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreads(
} }
} }
streams.push_back(new MergeTreeBlockInputStream( res.push_back(new MergeTreeBlockInputStream(
data.getFullPath() + part.data_part->name + '/', max_block_size, column_names, data, data.getFullPath() + part.data_part->name + '/', max_block_size, column_names, data,
part.data_part, ranges_to_get_from_part, use_uncompressed_cache, part.data_part, ranges_to_get_from_part, use_uncompressed_cache,
prewhere_actions, prewhere_column)); prewhere_actions, prewhere_column));
@ -435,18 +433,13 @@ BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreads(
for (const String & virt_column : virt_columns) for (const String & virt_column : virt_columns)
{ {
if (virt_column == "_part") if (virt_column == "_part")
streams.back() = new AddingConstColumnBlockInputStream<String>( res.back() = new AddingConstColumnBlockInputStream<String>(
streams.back(), new DataTypeString, part.data_part->name, "_part"); res.back(), new DataTypeString, part.data_part->name, "_part");
else if (virt_column == "_part_index") else if (virt_column == "_part_index")
streams.back() = new AddingConstColumnBlockInputStream<UInt64>( res.back() = new AddingConstColumnBlockInputStream<UInt64>(
streams.back(), new DataTypeUInt64, part.part_index_in_query, "_part_index"); res.back(), new DataTypeUInt64, part.part_index_in_query, "_part_index");
} }
} }
if (streams.size() == 1)
res.push_back(streams[0]);
else
res.push_back(new ConcatBlockInputStream(streams));
} }
if (!parts.empty()) if (!parts.empty())

6
dbms/src/Storages/StorageChunkMerger.cpp
View File

@ -12,7 +12,6 @@
#include <DB/Interpreters/executeQuery.h> #include <DB/Interpreters/executeQuery.h>
#include <DB/Interpreters/InterpreterDropQuery.h> #include <DB/Interpreters/InterpreterDropQuery.h>
#include <DB/DataStreams/ConcatBlockInputStream.h> #include <DB/DataStreams/ConcatBlockInputStream.h>
#include <DB/DataStreams/narrowBlockInputStreams.h>
#include <DB/DataStreams/AddingDefaultBlockInputStream.h> #include <DB/DataStreams/AddingDefaultBlockInputStream.h>
#include <DB/DataStreams/AddingConstColumnBlockInputStream.h> #include <DB/DataStreams/AddingConstColumnBlockInputStream.h>
#include <DB/Common/VirtualColumnUtils.h> #include <DB/Common/VirtualColumnUtils.h>
@ -210,11 +209,6 @@ BlockInputStreams StorageChunkMerger::read(
processed_stage = tmp_processed_stage; processed_stage = tmp_processed_stage;
} }
/** Если истчоников слишком много, то склеим их в threads источников.
*/
if (res.size() > threads)
res = narrowBlockInputStreams(res, threads);
return res; return res;
} }

6
dbms/src/Storages/StorageMerge.cpp
View File

@ -1,4 +1,3 @@
#include <DB/DataStreams/narrowBlockInputStreams.h>
#include <DB/DataStreams/AddingConstColumnBlockInputStream.h> #include <DB/DataStreams/AddingConstColumnBlockInputStream.h>
#include <DB/Storages/StorageMerge.h> #include <DB/Storages/StorageMerge.h>
#include <DB/Common/VirtualColumnUtils.h> #include <DB/Common/VirtualColumnUtils.h>
@ -173,11 +172,6 @@ BlockInputStreams StorageMerge::read(
processed_stage = tmp_processed_stage; processed_stage = tmp_processed_stage;
} }
/** Если истчоников слишком много, то склеим их в threads источников.
*/
if (res.size() > threads)
res = narrowBlockInputStreams(res, threads);
return res; return res;
} }