ClickHouse/dbms/src/Interpreters/ActionLocksManager.cpp

99 lines
2.6 KiB
C++
Raw Normal View History

#include "ActionLocksManager.h"
#include <Interpreters/Context.h>
#include <Databases/IDatabase.h>
#include <Storages/IStorage.h>
namespace DB
{
namespace ActionLocks
{
extern const StorageActionBlockType PartsMerge = 1;
extern const StorageActionBlockType PartsFetch = 2;
extern const StorageActionBlockType PartsSend = 3;
extern const StorageActionBlockType ReplicationQueue = 4;
2019-04-22 15:11:16 +00:00
extern const StorageActionBlockType DistributedSend = 5;
}
template <typename F>
inline void forEachTable(Context & context, F && f)
{
for (auto & elem : context.getDatabases())
for (auto iterator = elem.second->getIterator(context); iterator->isValid(); iterator->next())
f(iterator->table());
}
void ActionLocksManager::add(StorageActionBlockType action_type)
{
forEachTable(global_context, [&] (const StoragePtr & table)
{
ActionLock action_lock = table->getActionLock(action_type);
if (!action_lock.expired())
{
2019-01-02 06:44:36 +00:00
std::lock_guard lock(mutex);
storage_locks[table.get()][action_type] = std::move(action_lock);
}
});
}
void ActionLocksManager::add(const String & database_name, const String & table_name, StorageActionBlockType action_type)
{
if (auto table = global_context.tryGetTable(database_name, table_name))
{
ActionLock action_lock = table->getActionLock(action_type);
if (!action_lock.expired())
{
2019-01-02 06:44:36 +00:00
std::lock_guard lock(mutex);
storage_locks[table.get()][action_type] = std::move(action_lock);
}
}
}
void ActionLocksManager::remove(StorageActionBlockType action_type)
{
2019-01-02 06:44:36 +00:00
std::lock_guard lock(mutex);
for (auto & storage_elem : storage_locks)
storage_elem.second.erase(action_type);
}
void ActionLocksManager::remove(const String & database_name, const String & table_name, StorageActionBlockType action_type)
{
if (auto table = global_context.tryGetTable(database_name, table_name))
{
2019-01-02 06:44:36 +00:00
std::lock_guard lock(mutex);
if (storage_locks.count(table.get()))
storage_locks[table.get()].erase(action_type);
}
}
void ActionLocksManager::cleanExpired()
{
2019-01-02 06:44:36 +00:00
std::lock_guard lock(mutex);
2018-11-24 01:48:06 +00:00
for (auto it_storage = storage_locks.begin(); it_storage != storage_locks.end();)
{
auto & locks = it_storage->second;
2018-11-24 01:48:06 +00:00
for (auto it_lock = locks.begin(); it_lock != locks.end();)
{
if (it_lock->second.expired())
it_lock = locks.erase(it_lock);
else
++it_lock;
}
if (locks.empty())
it_storage = storage_locks.erase(it_storage);
else
++it_storage;
}
}
}