thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-12-13 01:41:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

remove stale loogs from asynchronous insertion queue

Browse Source
This commit is contained in:
Anton Popov 2021-08-30 16:37:27 +03:00
parent 7aec53bb8f
commit 8f60c4b8d2
2 changed files with 30 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/Interpreters/AsynchronousInsertionQueue.cpp
View File

@ -94,7 +94,6 @@ AsynchronousInsertQueue::AsynchronousInsertQueue(ContextMutablePtr context_, siz
, max_data_size(max_data_size_)
, busy_timeout(timeouts.busy)
, stale_timeout(timeouts.stale)
, lock(RWLockImpl::create())
, queue(new Queue)
, pool(pool_size)
, dump_by_first_update_thread(&AsynchronousInsertQueue::busyCheck, this)
@ -124,7 +123,7 @@ AsynchronousInsertQueue::~AsynchronousInsertQueue()
void AsynchronousInsertQueue::push(const ASTPtr & query, const Settings & settings, const String & query_id)
{
auto write_lock = lock->getLock(RWLockImpl::Write, String());
std::unique_lock lock(rwlock);
InsertQuery key{query, settings};
auto it = queue->find(key);
@ -165,20 +164,38 @@ void AsynchronousInsertQueue::busyCheck()
{
std::this_thread::sleep_for(timeout);
auto read_lock = lock->getLock(RWLockImpl::Read, String());
/// TODO: use priority queue instead of raw unsorted queue.
timeout = busy_timeout;
for (auto & [_, data] : *queue)
std::vector<InsertQuery> keys_to_remove;
{
std::unique_lock<std::mutex> data_lock(data->mutex);
std::shared_lock read_lock(rwlock);
auto lag = std::chrono::steady_clock::now() - data->first_update;
for (auto & [key, data] : *queue)
{
std::unique_lock<std::mutex> data_lock(data->mutex);
if (lag >= busy_timeout)
pool.scheduleOrThrowOnError([data = data, global_context = getContext()] { processData(data, global_context); });
else
timeout = std::min(timeout, std::chrono::ceil<std::chrono::seconds>(busy_timeout - lag));
auto lag = std::chrono::steady_clock::now() - data->first_update;
if (data->is_reset)
keys_to_remove.push_back(key);
else if (lag >= busy_timeout)
pool.scheduleOrThrowOnError([data = data, global_context = getContext()] { processData(data, global_context); });
else
timeout = std::min(timeout, std::chrono::ceil<std::chrono::seconds>(busy_timeout - lag));
}
}
if (!keys_to_remove.empty())
{
std::unique_lock write_lock(rwlock);
for (const auto & key : keys_to_remove)
{
auto it = queue->find(key);
if (it != queue->end() && it->second->is_reset)
queue->erase(it);
}
}
}
}
@ -189,7 +206,7 @@ void AsynchronousInsertQueue::staleCheck()
{
std::this_thread::sleep_for(stale_timeout);
auto read_lock = lock->getLock(RWLockImpl::Read, String());
std::shared_lock read_lock(rwlock);
for (auto & [_, data] : *queue)
{

2
src/Interpreters/AsynchronousInsertionQueue.h
View File

@ -65,7 +65,7 @@ class AsynchronousInsertQueue : public WithMutableContext
const size_t max_data_size; /// in bytes
const std::chrono::seconds busy_timeout, stale_timeout;
RWLock lock;
std::shared_mutex rwlock;
std::unique_ptr<Queue> queue;
std::atomic<bool> shutdown{false};