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
81
82
83
84
85
86
87
88
89
90
91
92
93
|
#pragma once
#include <Common/CurrentMetrics.h>
#include <memory>
#include <list>
#include <mutex>
#include <atomic>
namespace DB
{
/// Common code for background processes lists, like system.merges and system.replicated_fetches
/// Look at examples in MergeList, MovesList and ReplicatedFetchList
template <typename ListElement, typename Info>
class BackgroundProcessList;
template <typename ListElement, typename Info>
class BackgroundProcessListEntry
{
BackgroundProcessList<ListElement, Info> & list;
using container_t = std::list<ListElement>;
typename container_t::iterator it;
CurrentMetrics::Increment metric_increment;
public:
BackgroundProcessListEntry(const BackgroundProcessListEntry &) = delete;
BackgroundProcessListEntry & operator=(const BackgroundProcessListEntry &) = delete;
BackgroundProcessListEntry(BackgroundProcessListEntry &&) noexcept = default;
BackgroundProcessListEntry(BackgroundProcessList<ListElement, Info> & list_, const typename container_t::iterator it_, const CurrentMetrics::Metric & metric)
: list(list_), it{it_}, metric_increment{metric}
{
list.onEntryCreate(*this);
}
~BackgroundProcessListEntry()
{
std::lock_guard lock{list.mutex};
list.onEntryDestroy(*this);
list.entries.erase(it);
}
ListElement * operator->() { return &*it; }
const ListElement * operator->() const { return &*it; }
};
template <typename ListElement, typename Info>
class BackgroundProcessList
{
protected:
friend class BackgroundProcessListEntry<ListElement, Info>;
using container_t = std::list<ListElement>;
using info_container_t = std::list<Info>;
mutable std::mutex mutex;
container_t entries;
CurrentMetrics::Metric metric;
explicit BackgroundProcessList(const CurrentMetrics::Metric & metric_)
: metric(metric_)
{}
public:
using Entry = BackgroundProcessListEntry<ListElement, Info>;
using EntryPtr = std::unique_ptr<Entry>;
template <typename... Args>
EntryPtr insert(Args &&... args)
{
std::lock_guard lock{mutex};
auto entry = std::make_unique<Entry>(*this, entries.emplace(entries.end(), std::forward<Args>(args)...), metric);
return entry;
}
info_container_t get() const
{
std::lock_guard lock{mutex};
info_container_t res;
for (const auto & list_element : entries)
res.emplace_back(list_element.getInfo());
return res;
}
virtual void onEntryCreate(const Entry & /* entry */) {}
virtual void onEntryDestroy(const Entry & /* entry */) {}
virtual inline ~BackgroundProcessList() = default;
};
}
|
