thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-27 18:12:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Try fix

Browse Source
This commit is contained in:
avogar 2023-03-22 12:25:16 +00:00
parent b786f6fa3f
commit 783e3ab789
3 changed files with 20 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
src/QueryPipeline/RemoteQueryExecutor.cpp
View File

@ -218,23 +218,6 @@ static Block adaptBlockStructure(const Block & block, const Block & header)
void RemoteQueryExecutor::sendQuery(ClientInfo::QueryKind query_kind, AsyncCallback async_callback) void RemoteQueryExecutor::sendQuery(ClientInfo::QueryKind query_kind, AsyncCallback async_callback)
{ {
if (sent_query)
return;
connections = create_connections(async_callback);
AsyncCallbackSetter async_callback_setter(connections.get(), async_callback);
const auto & settings = context->getSettingsRef();
if (needToSkipUnavailableShard())
return;
/// Query could be cancelled during creating connections and this code can be executed
/// inside read_context->cancel() under was_cancelled_mutex, it can happen only when
/// was_cancelled = true (because it's set to true before calling read_context->cancel())
/// To avoid deadlock, we should check was_cancelled before locking was_cancelled_mutex.
if (was_cancelled)
return;
/// Query cannot be canceled in the middle of the send query, /// Query cannot be canceled in the middle of the send query,
/// since there are multiple packets: /// since there are multiple packets:
/// - Query /// - Query
@ -245,9 +228,26 @@ void RemoteQueryExecutor::sendQuery(ClientInfo::QueryKind query_kind, AsyncCallb
/// Unexpected packet Data received from client /// Unexpected packet Data received from client
/// ///
std::lock_guard guard(was_cancelled_mutex); std::lock_guard guard(was_cancelled_mutex);
sendQueryUnlocked(query_kind, async_callback);
}
void RemoteQueryExecutor::sendQueryUnlocked(ClientInfo::QueryKind query_kind, AsyncCallback async_callback)
{
if (sent_query || was_cancelled)
return;
connections = create_connections(async_callback);
AsyncCallbackSetter async_callback_setter(connections.get(), async_callback);
const auto & settings = context->getSettingsRef();
if (needToSkipUnavailableShard())
return;
established = true; established = true;
was_cancelled = false;
/// Query could be cancelled during creating connections. No need to send a query.
if (was_cancelled)
return;
auto timeouts = ConnectionTimeouts::getTCPTimeoutsWithFailover(settings); auto timeouts = ConnectionTimeouts::getTCPTimeoutsWithFailover(settings);
ClientInfo modified_client_info = context->getClientInfo(); ClientInfo modified_client_info = context->getClientInfo();

1
src/QueryPipeline/RemoteQueryExecutor.h
View File

@ -89,6 +89,7 @@ public:
/// But clickhouse-benchmark uses the same code, /// But clickhouse-benchmark uses the same code,
/// and it should pass INITIAL_QUERY. /// and it should pass INITIAL_QUERY.
void sendQuery(ClientInfo::QueryKind query_kind = ClientInfo::QueryKind::SECONDARY_QUERY, AsyncCallback async_callback = {}); void sendQuery(ClientInfo::QueryKind query_kind = ClientInfo::QueryKind::SECONDARY_QUERY, AsyncCallback async_callback = {});
void sendQueryUnlocked(ClientInfo::QueryKind query_kind = ClientInfo::QueryKind::SECONDARY_QUERY, AsyncCallback async_callback = {});
int sendQueryAsync(); int sendQueryAsync();

2
src/QueryPipeline/RemoteQueryExecutorReadContext.cpp
View File

@ -36,7 +36,7 @@ bool RemoteQueryExecutorReadContext::checkBeforeTaskResume()
void RemoteQueryExecutorReadContext::Task::run(AsyncCallback async_callback, ResumeCallback suspend_callback) void RemoteQueryExecutorReadContext::Task::run(AsyncCallback async_callback, ResumeCallback suspend_callback)
{ {
read_context.executor.sendQuery(ClientInfo::QueryKind::SECONDARY_QUERY, async_callback); read_context.executor.sendQueryUnlocked(ClientInfo::QueryKind::SECONDARY_QUERY, async_callback);
read_context.is_query_sent = true; read_context.is_query_sent = true;
if (read_context.suspend_when_query_sent) if (read_context.suspend_when_query_sent)