thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-29 21:20:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

slightly better

Browse Source
This commit is contained in:
Anton Popov 2021-08-31 05:35:36 +03:00
parent 36ac20681e
commit fb0790cf82
2 changed files with 14 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
src/Interpreters/AsynchronousInsertionQueue.cpp
View File

@ -124,6 +124,16 @@ AsynchronousInsertQueue::~AsynchronousInsertQueue()
}
}
void AsynchronousInsertQueue::scheduleProcessJob(const InsertQuery & key, InsertDataPtr data, ContextPtr global_context)
{
/// Wrap 'unique_ptr' with 'shared_ptr' to make this
/// lambda copyable and allow to save it to the thread pool.
pool.scheduleOrThrowOnError([=, data = std::make_shared<InsertDataPtr>(std::move(data))]
{
processData(std::move(key), std::move(*data), std::move(global_context));
});
}
void AsynchronousInsertQueue::push(const ASTPtr & query, const Settings & settings, const String & query_id)
{
auto read_buf = getReadBufferFromASTInsertQuery(query);
@ -171,14 +181,7 @@ void AsynchronousInsertQueue::push(const ASTPtr & query, const Settings & settin
LOG_INFO(log, "Queue size {} for query '{}'", data->size, queryToString(*query));
if (data->size > max_data_size)
{
pool.scheduleOrThrowOnError([key,
data = std::make_shared<InsertDataPtr>(std::move(data)),
global_context = getContext()]
{
processData(std::move(key), std::move(*data), global_context);
});
}
scheduleProcessJob(key, std::move(data), getContext());
}
void AsynchronousInsertQueue::waitForProcessingQuery(const String & query_id, const Milliseconds & timeout)
@ -223,14 +226,7 @@ void AsynchronousInsertQueue::busyCheck()
auto lag = std::chrono::steady_clock::now() - elem->data->first_update;
if (lag >= busy_timeout)
{
pool.scheduleOrThrowOnError([key = key,
data = std::make_shared<InsertDataPtr>(std::move(elem->data)),
global_context = getContext()]
{
processData(std::move(key), std::move(*data), global_context);
});
}
scheduleProcessJob(key, std::move(elem->data), getContext());
else
timeout = std::min(timeout, std::chrono::ceil<std::chrono::seconds>(busy_timeout - lag));
}
@ -252,14 +248,7 @@ void AsynchronousInsertQueue::staleCheck()
auto lag = std::chrono::steady_clock::now() - elem->data->last_update;
if (lag >= stale_timeout)
{
pool.scheduleOrThrowOnError([key = key,
data = std::make_shared<InsertDataPtr>(std::move(elem->data)),
global_context = getContext()]
{
processData(std::move(key), std::move(*data), global_context);
});
}
scheduleProcessJob(key, std::move(elem->data), getContext());
}
}
}

1
src/Interpreters/AsynchronousInsertionQueue.h
View File

@ -124,6 +124,7 @@ private:
void staleCheck();
void cleanup();
void scheduleProcessJob(const InsertQuery & key, InsertDataPtr data, ContextPtr global_context);
static void processData(InsertQuery key, InsertDataPtr data, ContextPtr global_context);
};