thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-26 01:22:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update TCPHandler.

Browse Source
This commit is contained in:
Nikolai Kochetov 2020-01-23 13:04:18 +03:00
parent 0e906b29e1
commit f431b10e38
2 changed files with 52 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
dbms/programs/server/TCPHandler.cpp
View File

@ -591,11 +591,9 @@ void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)
}
});
/// Wait in case of exception. Delete pipeline to release memory.
/// Wait in case of exception happened outside of pool.
SCOPE_EXIT(
/// Clear queue in case if somebody is waiting lazy_format to push.
lazy_format->finish();
lazy_format->clearQueue();
try
{
@ -604,26 +602,19 @@ void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)
catch (...)
{
/// If exception was thrown during pipeline execution, skip it while processing other exception.
tryLogCurrentException(log);
}
/// pipeline = QueryPipeline()
);
while (true)
{
Block block;
while (true)
while (!lazy_format->isFinished() && !exception)
{
if (isQueryCancelled())
{
/// A packet was received requesting to stop execution of the request.
executor->cancel();
break;
}
else
{
if (after_send_progress.elapsed() / 1000 >= query_context->getSettingsRef().interactive_delay)
{
/// Some time passed and there is a progress.
@ -633,19 +624,16 @@ void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)
sendLogs();
if ((block = lazy_format->getBlock(query_context->getSettingsRef().interactive_delay / 1000)))
break;
if (lazy_format->isFinished())
break;
if (exception)
if (auto block = lazy_format->getBlock(query_context->getSettingsRef().interactive_delay / 1000))
{
if (!state.io.null_format)
sendData(block);
}
}
/// Finish lazy_format before waiting. Otherwise some thread may write into it, and waiting will lock.
lazy_format->finish();
pool.wait();
break;
}
}
}
/** If data has run out, we will send the profiling data and total values to
* the last zero block to be able to use
@ -654,9 +642,8 @@ void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)
* because we have not read all the data yet,
* and there could be ongoing calculations in other threads at the same time.
*/
if (!block && !isQueryCancelled())
if (!isQueryCancelled())
{
pool.wait();
pipeline.finalize();
sendTotals(lazy_format->getTotals());
@ -666,10 +653,7 @@ void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)
sendLogs();
}
sendData(block);
if (!block)
break;
}
sendData({});
}
state.io.onFinish();

8
dbms/src/Processors/Formats/LazyOutputFormat.h
View File

@ -26,8 +26,12 @@ public:
void setRowsBeforeLimit(size_t rows_before_limit) override;
void finish() { finished_processing = true; }
void clearQueue() { queue.clear(); }
void finish()
{
finished_processing = true;
/// Clear queue in case if somebody is waiting lazy_format to push.
queue.clear();
}
protected:
void consume(Chunk chunk) override