ClickHouse/dbms/include/DB/DataStreams/AsynchronousBlockInputStream.h

140 lines
3.4 KiB
C
Raw Normal View History

2011-11-28 05:48:52 +00:00
#pragma once
2012-10-20 05:36:32 +00:00
#include <statdaemons/threadpool.hpp>
#include <Poco/Event.h>
2011-11-28 05:48:52 +00:00
2011-11-28 06:22:25 +00:00
#include <DB/DataStreams/IProfilingBlockInputStream.h>
2011-11-28 05:48:52 +00:00
namespace DB
{
2012-10-20 02:10:47 +00:00
/** Выполняет другой BlockInputStream в отдельном потоке.
* Это служит для двух целей:
* 1. Позволяет сделать так, чтобы разные стадии конвеьера выполнения запроса работали параллельно.
* 2. Позволяет не ждать до того, как данные будут готовы, а периодически проверять их готовность без блокировки.
* Это нужно, например, чтобы можно было во время ожидания проверить, не пришёл ли по сети пакет с просьбой прервать выполнение запроса.
* Также это позволяет выполнить несколько запросов одновременно.
*/
class AsynchronousBlockInputStream : public IProfilingBlockInputStream
{
public:
2013-05-04 04:05:15 +00:00
AsynchronousBlockInputStream(BlockInputStreamPtr in_) : pool(1), started(false)
2012-10-20 02:10:47 +00:00
{
2013-05-04 04:05:15 +00:00
children.push_back(in_);
2012-10-20 02:10:47 +00:00
}
String getName() const override { return "AsynchronousBlockInputStream"; }
2012-10-20 05:36:32 +00:00
String getID() const override
{
std::stringstream res;
2013-05-04 05:20:07 +00:00
res << "Asynchronous(" << children.back()->getID() << ")";
return res.str();
}
void readPrefix() override
{
children.back()->readPrefix();
next();
started = true;
}
void readSuffix() override
{
if (started)
{
pool.wait();
if (exception)
exception->rethrow();
children.back()->readSuffix();
started = false;
}
}
2011-11-28 05:48:52 +00:00
2012-10-20 02:10:47 +00:00
/** Ждать готовность данных не более заданного таймаута. Запустить получение данных, если нужно.
2012-10-20 05:36:32 +00:00
* Если функция вернула true - данные готовы и можно делать read(); нельзя вызвать функцию сразу ещё раз.
2012-10-20 02:10:47 +00:00
*/
bool poll(UInt64 milliseconds)
{
2012-10-20 05:36:32 +00:00
if (!started)
{
next();
started = true;
}
return ready.tryWait(milliseconds);
2011-11-28 05:48:52 +00:00
}
~AsynchronousBlockInputStream() override
2012-03-05 02:34:20 +00:00
{
if (started)
2012-10-20 05:36:32 +00:00
pool.wait();
2012-03-05 02:34:20 +00:00
}
2011-11-28 05:48:52 +00:00
protected:
2012-10-20 05:36:32 +00:00
boost::threadpool::pool pool;
Poco::Event ready;
bool started;
2012-10-20 02:10:47 +00:00
2012-10-20 05:36:32 +00:00
Block block;
ExceptionPtr exception;
2012-10-20 02:10:47 +00:00
2014-07-16 02:06:58 +00:00
Block readImpl() override
2011-11-28 05:48:52 +00:00
{
2012-10-20 05:36:32 +00:00
/// Если вычислений ещё не было - вычислим первый блок синхронно
2012-10-20 02:10:47 +00:00
if (!started)
2011-11-28 05:48:52 +00:00
{
calculate(current_memory_tracker);
2012-10-20 02:35:44 +00:00
started = true;
2011-11-28 05:48:52 +00:00
}
2012-10-20 05:36:32 +00:00
else /// Если вычисления уже идут - подождём результата
pool.wait();
if (exception)
exception->rethrow();
Block res = block;
if (!res)
return res;
/// Запустим вычисления следующего блока
block = Block();
next();
return res;
2012-10-20 02:10:47 +00:00
}
2014-07-16 02:06:58 +00:00
2012-10-20 02:10:47 +00:00
2012-10-20 05:36:32 +00:00
void next()
{
ready.reset();
pool.schedule(std::bind(&AsynchronousBlockInputStream::calculate, this, current_memory_tracker));
2012-10-20 05:36:32 +00:00
}
2014-07-16 02:06:58 +00:00
2012-10-20 02:10:47 +00:00
/// Вычисления, которые могут выполняться в отдельном потоке
void calculate(MemoryTracker * memory_tracker)
2012-10-20 02:10:47 +00:00
{
current_memory_tracker = memory_tracker;
2012-10-20 05:36:32 +00:00
try
2011-11-28 05:48:52 +00:00
{
2013-05-04 05:20:07 +00:00
block = children.back()->read();
2011-11-28 05:48:52 +00:00
}
2012-10-20 05:36:32 +00:00
catch (...)
{
exception = cloneCurrentException();
2011-11-28 05:48:52 +00:00
}
2012-10-20 02:10:47 +00:00
2012-10-20 05:36:32 +00:00
ready.set();
}
2011-11-28 05:48:52 +00:00
};
}