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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
#include <Storages/MergeTree/BackgroundJobsAssignee.h>
#include <Storages/MergeTree/MergeTreeData.h>
#include <Common/CurrentMetrics.h>
#include <Common/randomSeed.h>
#include <Interpreters/Context.h>
#include <pcg_random.hpp>
#include <random>
namespace DB
{
BackgroundJobsAssignee::BackgroundJobsAssignee(MergeTreeData & data_, BackgroundJobsAssignee::Type type_, ContextPtr global_context_)
: WithContext(global_context_)
, data(data_)
, sleep_settings(global_context_->getBackgroundMoveTaskSchedulingSettings())
, rng(randomSeed())
, type(type_)
{
}
void BackgroundJobsAssignee::trigger()
{
std::lock_guard lock(holder_mutex);
if (!holder)
return;
/// Do not reset backoff factor if some task has appeared,
/// but decrease it exponentially on every new task.
no_work_done_count /= 2;
/// We have background jobs, schedule task as soon as possible
holder->schedule();
}
void BackgroundJobsAssignee::postpone()
{
std::lock_guard lock(holder_mutex);
if (!holder)
return;
no_work_done_count += 1;
double random_addition = std::uniform_real_distribution<double>(0, sleep_settings.task_sleep_seconds_when_no_work_random_part)(rng);
size_t next_time_to_execute = static_cast<size_t>(
1000 * (std::min(
sleep_settings.task_sleep_seconds_when_no_work_max,
sleep_settings.thread_sleep_seconds_if_nothing_to_do * std::pow(sleep_settings.task_sleep_seconds_when_no_work_multiplier, no_work_done_count))
+ random_addition));
holder->scheduleAfter(next_time_to_execute, false);
}
bool BackgroundJobsAssignee::scheduleMergeMutateTask(ExecutableTaskPtr merge_task)
{
bool res = getContext()->getMergeMutateExecutor()->trySchedule(merge_task);
res ? trigger() : postpone();
return res;
}
void BackgroundJobsAssignee::scheduleFetchTask(ExecutableTaskPtr fetch_task)
{
bool res = getContext()->getFetchesExecutor()->trySchedule(fetch_task);
res ? trigger() : postpone();
}
void BackgroundJobsAssignee::scheduleMoveTask(ExecutableTaskPtr move_task)
{
bool res = getContext()->getMovesExecutor()->trySchedule(move_task);
res ? trigger() : postpone();
}
void BackgroundJobsAssignee::scheduleCommonTask(ExecutableTaskPtr common_task, bool need_trigger)
{
bool res = getContext()->getCommonExecutor()->trySchedule(common_task) && need_trigger;
res ? trigger() : postpone();
}
String BackgroundJobsAssignee::toString(Type type)
{
switch (type)
{
case Type::DataProcessing:
return "DataProcessing";
case Type::Moving:
return "Moving";
}
UNREACHABLE();
}
void BackgroundJobsAssignee::start()
{
std::lock_guard lock(holder_mutex);
if (!holder)
holder = getContext()->getSchedulePool().createTask("BackgroundJobsAssignee:" + toString(type), [this]{ threadFunc(); });
holder->activateAndSchedule();
}
void BackgroundJobsAssignee::finish()
{
/// No lock here, because scheduled tasks could call trigger method
if (holder)
{
holder->deactivate();
auto storage_id = data.getStorageID();
getContext()->getMovesExecutor()->removeTasksCorrespondingToStorage(storage_id);
getContext()->getFetchesExecutor()->removeTasksCorrespondingToStorage(storage_id);
getContext()->getMergeMutateExecutor()->removeTasksCorrespondingToStorage(storage_id);
getContext()->getCommonExecutor()->removeTasksCorrespondingToStorage(storage_id);
}
}
void BackgroundJobsAssignee::threadFunc()
try
{
bool succeed = false;
switch (type)
{
case Type::DataProcessing:
succeed = data.scheduleDataProcessingJob(*this);
break;
case Type::Moving:
succeed = data.scheduleDataMovingJob(*this);
break;
}
if (!succeed)
postpone();
}
catch (...) /// Catch any exception to avoid thread termination.
{
tryLogCurrentException(__PRETTY_FUNCTION__);
postpone();
}
BackgroundJobsAssignee::~BackgroundJobsAssignee()
{
try
{
finish();
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
}
}
}
|
