thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-21 23:21:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #22858 from vitlibar/fix-hanging-in-temporary-live-view-cleaner

Browse Source 
Fix hanging in TemporaryLiveViewCleaner
This commit is contained in:
alexey-milovidov 2021-04-09 09:25:38 +03:00 committed by GitHub
commit b02a82250b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 28 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/Interpreters/DatabaseCatalog.cpp
View File

@ -152,7 +152,7 @@ void DatabaseCatalog::loadDatabases()
/// Another background thread which drops temporary LiveViews.
/// We should start it after loadMarkedAsDroppedTables() to avoid race condition.
TemporaryLiveViewCleaner::instance().startupIfNecessary();
TemporaryLiveViewCleaner::instance().startup();
}
void DatabaseCatalog::shutdownImpl()

69
src/Storages/LiveView/TemporaryLiveViewCleaner.cpp
View File

@ -48,38 +48,13 @@ void TemporaryLiveViewCleaner::init(Context & global_context_)
the_instance.reset(new TemporaryLiveViewCleaner(global_context_));
}
void TemporaryLiveViewCleaner::startupIfNecessary()
void TemporaryLiveViewCleaner::startup()
{
background_thread_can_start = true;
std::lock_guard lock{mutex};
if (background_thread_should_exit)
return;
if (!views.empty())
startupIfNecessaryImpl(lock);
else
can_start_background_thread = true;
}
void TemporaryLiveViewCleaner::startupIfNecessaryImpl(const std::lock_guard<std::mutex> &)
{
/// If views.empty() the background thread isn't running or it's going to stop right now.
/// If can_start_background_thread is false, then the thread has not been started previously.
bool background_thread_is_running;
if (can_start_background_thread)
{
background_thread_is_running = !views.empty();
}
else
{
can_start_background_thread = true;
background_thread_is_running = false;
}
if (!background_thread_is_running)
{
if (background_thread.joinable())
background_thread.join();
background_thread = ThreadFromGlobalPool{&TemporaryLiveViewCleaner::backgroundThreadFunc, this};
}
startBackgroundThread();
}
void TemporaryLiveViewCleaner::shutdown()
@ -87,13 +62,11 @@ void TemporaryLiveViewCleaner::shutdown()
the_instance.reset();
}
TemporaryLiveViewCleaner::TemporaryLiveViewCleaner(Context & global_context_)
: global_context(global_context_)
{
}
TemporaryLiveViewCleaner::~TemporaryLiveViewCleaner()
{
stopBackgroundThread();
@ -108,27 +81,29 @@ void TemporaryLiveViewCleaner::addView(const std::shared_ptr<StorageLiveView> &
auto current_time = std::chrono::system_clock::now();
auto time_of_next_check = current_time + view->getTimeout();
std::lock_guard lock{mutex};
if (background_thread_should_exit)
return;
if (can_start_background_thread)
startupIfNecessaryImpl(lock);
/// Keep the vector `views` sorted by time of next check.
StorageAndTimeOfCheck storage_and_time_of_check{view, time_of_next_check};
std::lock_guard lock{mutex};
views.insert(std::upper_bound(views.begin(), views.end(), storage_and_time_of_check), storage_and_time_of_check);
background_thread_wake_up.notify_one();
if (background_thread_can_start)
{
startBackgroundThread();
background_thread_wake_up.notify_one();
}
}
void TemporaryLiveViewCleaner::backgroundThreadFunc()
{
std::unique_lock lock{mutex};
while (!background_thread_should_exit && !views.empty())
while (!background_thread_should_exit)
{
background_thread_wake_up.wait_until(lock, views.front().time_of_check);
if (views.empty())
background_thread_wake_up.wait(lock);
else
background_thread_wake_up.wait_until(lock, views.front().time_of_check);
if (background_thread_should_exit)
break;
@ -173,14 +148,18 @@ void TemporaryLiveViewCleaner::backgroundThreadFunc()
}
void TemporaryLiveViewCleaner::startBackgroundThread()
{
if (!background_thread.joinable() && background_thread_can_start && !background_thread_should_exit)
background_thread = ThreadFromGlobalPool{&TemporaryLiveViewCleaner::backgroundThreadFunc, this};
}
void TemporaryLiveViewCleaner::stopBackgroundThread()
{
background_thread_should_exit = true;
background_thread_wake_up.notify_one();
if (background_thread.joinable())
{
background_thread_should_exit = true;
background_thread_wake_up.notify_one();
background_thread.join();
}
}
}

6
src/Storages/LiveView/TemporaryLiveViewCleaner.h
View File

@ -23,8 +23,7 @@ public:
static void init(Context & global_context_);
static void shutdown();
void startupIfNecessary();
void startupIfNecessaryImpl(const std::lock_guard<std::mutex> &);
void startup();
private:
friend std::unique_ptr<TemporaryLiveViewCleaner>::deleter_type;
@ -33,6 +32,7 @@ private:
~TemporaryLiveViewCleaner();
void backgroundThreadFunc();
void startBackgroundThread();
void stopBackgroundThread();
struct StorageAndTimeOfCheck
@ -47,7 +47,7 @@ private:
std::mutex mutex;
std::vector<StorageAndTimeOfCheck> views;
ThreadFromGlobalPool background_thread;
bool can_start_background_thread = false;
std::atomic<bool> background_thread_can_start = false;
std::atomic<bool> background_thread_should_exit = false;
std::condition_variable background_thread_wake_up;
};