2012-01-10 22:11:51 +00:00
|
|
|
|
#pragma once
|
|
|
|
|
|
|
|
|
|
#include <Poco/Semaphore.h>
|
2012-09-03 19:52:57 +00:00
|
|
|
|
#include <Poco/ThreadPool.h>
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
|
|
|
|
#include <statdaemons/threadpool.hpp>
|
|
|
|
|
|
|
|
|
|
#include <DB/DataStreams/IProfilingBlockInputStream.h>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
|
{
|
|
|
|
|
|
|
|
|
|
using Poco::SharedPtr;
|
|
|
|
|
|
|
|
|
|
|
2012-09-03 19:52:57 +00:00
|
|
|
|
|
2012-01-10 22:11:51 +00:00
|
|
|
|
/** Объединяет несколько источников в один.
|
|
|
|
|
* Блоки из разных источников перемежаются друг с другом произвольным образом.
|
|
|
|
|
* Можно указать количество потоков (max_threads),
|
|
|
|
|
* в которых будет выполняться получение данных из разных источников.
|
|
|
|
|
*/
|
|
|
|
|
class UnionBlockInputStream : public IProfilingBlockInputStream
|
|
|
|
|
{
|
2012-09-04 19:57:17 +00:00
|
|
|
|
class Thread;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
public:
|
|
|
|
|
UnionBlockInputStream(BlockInputStreams inputs_, unsigned max_threads_ = 1)
|
|
|
|
|
: max_threads(std::min(inputs_.size(), static_cast<size_t>(max_threads_))),
|
2012-09-03 19:52:57 +00:00
|
|
|
|
pool(max_threads, max_threads),
|
2012-01-10 22:11:51 +00:00
|
|
|
|
ready_any(0, inputs_.size())
|
|
|
|
|
{
|
|
|
|
|
children.insert(children.end(), inputs_.begin(), inputs_.end());
|
2012-06-24 23:17:06 +00:00
|
|
|
|
|
2012-01-10 22:11:51 +00:00
|
|
|
|
for (size_t i = 0; i < inputs_.size(); ++i)
|
2012-06-24 23:17:06 +00:00
|
|
|
|
{
|
2012-09-04 19:57:17 +00:00
|
|
|
|
inputs_data.push_back(InputData());
|
|
|
|
|
inputs_data.back().in = inputs_[i];
|
|
|
|
|
inputs_data.back().i = i;
|
2012-06-24 23:17:06 +00:00
|
|
|
|
}
|
2012-01-10 22:11:51 +00:00
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
Block readImpl()
|
|
|
|
|
{
|
|
|
|
|
Block res;
|
|
|
|
|
|
2012-06-22 18:27:40 +00:00
|
|
|
|
while (1)
|
2012-01-10 22:11:51 +00:00
|
|
|
|
{
|
2012-06-22 18:27:40 +00:00
|
|
|
|
{
|
|
|
|
|
Poco::ScopedLock<Poco::FastMutex> lock(mutex);
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
2012-09-04 19:57:17 +00:00
|
|
|
|
if (inputs_data.empty())
|
2012-06-22 18:27:40 +00:00
|
|
|
|
return res;
|
2012-03-05 02:34:20 +00:00
|
|
|
|
|
2012-09-03 19:52:57 +00:00
|
|
|
|
ssize_t max_threads_to_start = pool.available();
|
2012-06-25 00:17:19 +00:00
|
|
|
|
|
|
|
|
|
if (max_threads_to_start > 0)
|
2012-01-10 22:11:51 +00:00
|
|
|
|
{
|
2012-06-24 23:17:06 +00:00
|
|
|
|
/// Запустим вычисления для как можно большего количества источников, которые ещё ни разу не брались
|
2012-06-25 03:56:45 +00:00
|
|
|
|
// std::cerr << "Starting initial threads" << std::endl;
|
2012-06-24 23:17:06 +00:00
|
|
|
|
|
2012-06-25 00:17:19 +00:00
|
|
|
|
ssize_t started_threads = 0;
|
2012-09-04 19:57:17 +00:00
|
|
|
|
InputsData::iterator it = inputs_data.begin();
|
|
|
|
|
while (it != inputs_data.end() && 0 == it->count)
|
2012-06-22 18:27:40 +00:00
|
|
|
|
{
|
2012-06-25 03:56:45 +00:00
|
|
|
|
// std::cerr << "Scheduling initial " << it->i << std::endl;
|
2012-06-24 23:17:06 +00:00
|
|
|
|
++it->count;
|
2012-06-22 18:27:40 +00:00
|
|
|
|
++started_threads;
|
2012-06-25 00:17:19 +00:00
|
|
|
|
|
|
|
|
|
/// Переносим этот источник в конец списка
|
2012-09-04 19:57:17 +00:00
|
|
|
|
inputs_data.push_back(*it);
|
|
|
|
|
inputs_data.erase(it++);
|
2012-09-04 20:19:25 +00:00
|
|
|
|
|
|
|
|
|
inputs_data.back().thread = new Thread(*this, inputs_data.back());
|
|
|
|
|
pool.start(*inputs_data.back().thread);
|
2012-06-22 18:27:40 +00:00
|
|
|
|
|
2012-06-24 23:17:06 +00:00
|
|
|
|
if (started_threads == max_threads_to_start)
|
2012-06-22 18:27:40 +00:00
|
|
|
|
break;
|
|
|
|
|
}
|
2012-01-10 22:11:51 +00:00
|
|
|
|
}
|
|
|
|
|
}
|
2012-06-24 23:17:06 +00:00
|
|
|
|
|
2012-06-25 02:52:51 +00:00
|
|
|
|
// std::cerr << "Waiting for one thread to finish" << std::endl;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
ready_any.wait();
|
|
|
|
|
|
|
|
|
|
{
|
|
|
|
|
Poco::ScopedLock<Poco::FastMutex> lock(mutex);
|
2012-03-05 02:34:20 +00:00
|
|
|
|
|
2012-06-25 02:52:51 +00:00
|
|
|
|
// std::cerr << std::endl << "pool.pending: " << pool.pending() << ", pool.active: " << pool.active() << ", pool.size: " << pool.size() << std::endl;
|
2012-06-24 23:17:06 +00:00
|
|
|
|
|
2012-09-04 19:57:17 +00:00
|
|
|
|
if (inputs_data.empty())
|
2012-01-10 22:11:51 +00:00
|
|
|
|
return res;
|
|
|
|
|
|
2012-06-25 02:52:51 +00:00
|
|
|
|
// std::cerr << "Searching for first ready block" << std::endl;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
|
|
|
|
/** Найдём и вернём готовый непустой блок, если такой есть.
|
|
|
|
|
* При чём, выберем блок из источника, из которого было получено меньше всего блоков.
|
|
|
|
|
*/
|
2012-09-04 19:57:17 +00:00
|
|
|
|
InputsData::iterator it = inputs_data.begin();
|
|
|
|
|
while (it != inputs_data.end())
|
2012-01-10 22:11:51 +00:00
|
|
|
|
{
|
2012-06-24 23:17:06 +00:00
|
|
|
|
if (it->exception)
|
|
|
|
|
it->exception->rethrow();
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
2012-06-24 23:17:06 +00:00
|
|
|
|
if (it->ready)
|
2012-01-10 22:11:51 +00:00
|
|
|
|
{
|
2012-06-24 23:17:06 +00:00
|
|
|
|
if (!it->block)
|
2012-09-04 19:57:17 +00:00
|
|
|
|
inputs_data.erase(it++);
|
2012-06-24 23:17:06 +00:00
|
|
|
|
else
|
|
|
|
|
break;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
}
|
2012-06-24 23:17:06 +00:00
|
|
|
|
else
|
|
|
|
|
++it;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
}
|
|
|
|
|
|
2012-09-04 19:57:17 +00:00
|
|
|
|
if (it == inputs_data.end())
|
2012-03-05 02:34:20 +00:00
|
|
|
|
{
|
2012-06-25 02:52:51 +00:00
|
|
|
|
// std::cerr << "Continue" << std::endl;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
continue;
|
2012-03-05 02:34:20 +00:00
|
|
|
|
}
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
2012-06-25 02:52:51 +00:00
|
|
|
|
// std::cerr << "Found block " << it->i << std::endl;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
2012-06-24 23:17:06 +00:00
|
|
|
|
res = it->block;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
|
|
|
|
/// Запустим получение следующего блока
|
2012-06-24 23:17:06 +00:00
|
|
|
|
it->reset();
|
2012-06-25 02:52:51 +00:00
|
|
|
|
// std::cerr << "Scheduling again " << it->i << std::endl;
|
2012-06-24 23:17:06 +00:00
|
|
|
|
++it->count;
|
2012-09-04 19:57:17 +00:00
|
|
|
|
it->thread = new Thread(*this, *it);
|
|
|
|
|
pool.start(*it->thread);
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
|
|
|
|
return res;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
String getName() const { return "UnionBlockInputStream"; }
|
|
|
|
|
|
|
|
|
|
BlockInputStreamPtr clone() { return new UnionBlockInputStream(children, max_threads); }
|
|
|
|
|
|
2012-06-22 18:16:47 +00:00
|
|
|
|
~UnionBlockInputStream()
|
2012-03-05 00:09:41 +00:00
|
|
|
|
{
|
2012-09-03 19:52:57 +00:00
|
|
|
|
pool.joinAll();
|
2012-03-05 00:09:41 +00:00
|
|
|
|
}
|
|
|
|
|
|
2012-01-10 22:11:51 +00:00
|
|
|
|
private:
|
|
|
|
|
unsigned max_threads;
|
|
|
|
|
|
2012-09-04 19:57:17 +00:00
|
|
|
|
/** Будем использовать Poco::ThreadPool вместо boost::threadpool.
|
|
|
|
|
* Последний неудобен тем, что в нём не совсем так как надо узнаётся количество свободных потоков.
|
|
|
|
|
*/
|
2012-09-03 19:52:57 +00:00
|
|
|
|
Poco::ThreadPool pool;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
|
|
|
|
/// Данные отдельного источника
|
2012-09-04 19:57:17 +00:00
|
|
|
|
struct InputData
|
2012-01-10 22:11:51 +00:00
|
|
|
|
{
|
2012-09-04 19:57:17 +00:00
|
|
|
|
SharedPtr<Thread> thread;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
BlockInputStreamPtr in;
|
|
|
|
|
unsigned count; /// Сколько блоков было вычислено
|
|
|
|
|
bool ready; /// Блок уже вычислен
|
|
|
|
|
Block block;
|
2012-03-30 19:56:18 +00:00
|
|
|
|
ExceptionPtr exception;
|
2012-06-24 23:17:06 +00:00
|
|
|
|
size_t i; /// Порядковый номер источника.
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
|
|
|
|
void reset()
|
|
|
|
|
{
|
|
|
|
|
ready = false;
|
|
|
|
|
block = Block();
|
|
|
|
|
exception = NULL;
|
2012-09-04 19:57:17 +00:00
|
|
|
|
thread = NULL;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
}
|
|
|
|
|
|
2012-09-04 19:57:17 +00:00
|
|
|
|
InputData() : count(0), ready(false), i(0) {}
|
2012-01-10 22:11:51 +00:00
|
|
|
|
};
|
|
|
|
|
|
2012-09-04 19:57:17 +00:00
|
|
|
|
class Thread : public Poco::Runnable
|
2012-01-10 22:11:51 +00:00
|
|
|
|
{
|
2012-09-04 19:57:17 +00:00
|
|
|
|
public:
|
|
|
|
|
Thread(UnionBlockInputStream & parent_, InputData & data_) : parent(parent_), data(data_) {}
|
|
|
|
|
|
|
|
|
|
void run()
|
2012-01-10 22:11:51 +00:00
|
|
|
|
{
|
2012-09-04 19:57:17 +00:00
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
Block block = data.in->read();
|
2012-01-10 22:11:51 +00:00
|
|
|
|
|
2012-09-04 19:57:17 +00:00
|
|
|
|
{
|
|
|
|
|
Poco::ScopedLock<Poco::FastMutex> lock(parent.mutex);
|
|
|
|
|
data.ready = true;
|
|
|
|
|
data.block = block;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
catch (const Exception & e)
|
2012-01-10 22:11:51 +00:00
|
|
|
|
{
|
2012-09-04 19:57:17 +00:00
|
|
|
|
data.exception = e.clone();
|
2012-01-10 22:11:51 +00:00
|
|
|
|
}
|
2012-09-04 19:57:17 +00:00
|
|
|
|
catch (const Poco::Exception & e)
|
|
|
|
|
{
|
|
|
|
|
data.exception = e.clone();
|
|
|
|
|
}
|
|
|
|
|
catch (const std::exception & e)
|
|
|
|
|
{
|
|
|
|
|
data.exception = new Exception(e.what(), ErrorCodes::STD_EXCEPTION);
|
|
|
|
|
}
|
|
|
|
|
catch (...)
|
|
|
|
|
{
|
|
|
|
|
data.exception = new Exception("Unknown exception", ErrorCodes::UNKNOWN_EXCEPTION);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
parent.ready_any.set();
|
2012-01-10 22:11:51 +00:00
|
|
|
|
}
|
2012-09-03 19:52:57 +00:00
|
|
|
|
|
2012-09-04 19:57:17 +00:00
|
|
|
|
private:
|
|
|
|
|
UnionBlockInputStream & parent;
|
|
|
|
|
InputData & data;
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
/// Список упорядочен по количеству полученных из источника блоков.
|
|
|
|
|
typedef std::list<InputData> InputsData;
|
|
|
|
|
InputsData inputs_data;
|
|
|
|
|
Poco::FastMutex mutex;
|
|
|
|
|
Poco::Semaphore ready_any;
|
2012-01-10 22:11:51 +00:00
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
}
|