blob: d7480d45524fe9bdccc19ed5522eb9d566e52574 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
#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;
extern const StorageActionBlockType DistributedSend = 5;
extern const StorageActionBlockType PartsTTLMerge = 6;
extern const StorageActionBlockType PartsMove = 7;
extern const StorageActionBlockType PullReplicationLog = 8;
}
ActionLocksManager::ActionLocksManager(ContextPtr context_) : WithContext(context_->getGlobalContext())
{
}
void ActionLocksManager::add(const StorageID & table_id, StorageActionBlockType action_type)
{
if (auto table = DatabaseCatalog::instance().tryGetTable(table_id, getContext()))
add(table, action_type);
}
void ActionLocksManager::add(const StoragePtr & table, StorageActionBlockType action_type)
{
ActionLock action_lock = table->getActionLock(action_type);
if (!action_lock.expired())
{
std::lock_guard lock(mutex);
storage_locks[table.get()][action_type] = std::move(action_lock);
}
}
void ActionLocksManager::remove(const StorageID & table_id, StorageActionBlockType action_type)
{
if (auto table = DatabaseCatalog::instance().tryGetTable(table_id, getContext()))
remove(table, action_type);
}
void ActionLocksManager::remove(const StoragePtr & table, StorageActionBlockType action_type)
{
std::lock_guard lock(mutex);
if (storage_locks.contains(table.get()))
storage_locks[table.get()].erase(action_type);
}
void ActionLocksManager::cleanExpired()
{
std::lock_guard lock(mutex);
for (auto it_storage = storage_locks.begin(); it_storage != storage_locks.end();)
{
auto & locks = it_storage->second;
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;
}
}
}
|
