1 | #include <Dictionaries/Embedded/RegionsHierarchies.h> |
2 | #include <Dictionaries/Embedded/RegionsNames.h> |
3 | #include <Dictionaries/Embedded/GeoDictionariesLoader.h> |
4 | #include <Interpreters/Context.h> |
5 | #include <Interpreters/EmbeddedDictionaries.h> |
6 | #include <Common/setThreadName.h> |
7 | #include <Common/Exception.h> |
8 | #include <common/logger_useful.h> |
9 | #include <Poco/Util/Application.h> |
10 | |
11 | |
12 | namespace DB |
13 | { |
14 | |
15 | namespace ErrorCodes |
16 | { |
17 | extern const int UNFINISHED; |
18 | } |
19 | |
20 | void EmbeddedDictionaries::handleException(const bool throw_on_error) const |
21 | { |
22 | const auto exception_ptr = std::current_exception(); |
23 | |
24 | tryLogCurrentException(log, "Cannot load dictionary! You must resolve this manually." ); |
25 | |
26 | if (throw_on_error) |
27 | std::rethrow_exception(exception_ptr); |
28 | } |
29 | |
30 | |
31 | template <typename Dictionary> |
32 | bool EmbeddedDictionaries::reloadDictionary( |
33 | MultiVersion<Dictionary> & dictionary, |
34 | DictionaryReloader<Dictionary> reload_dictionary, |
35 | const bool throw_on_error, |
36 | const bool force_reload) |
37 | { |
38 | const auto & config = context.getConfigRef(); |
39 | |
40 | bool not_initialized = dictionary.get() == nullptr; |
41 | |
42 | if (force_reload || !is_fast_start_stage || not_initialized) |
43 | { |
44 | try |
45 | { |
46 | auto new_dictionary = reload_dictionary(config); |
47 | if (new_dictionary) |
48 | dictionary.set(std::move(new_dictionary)); |
49 | } |
50 | catch (...) |
51 | { |
52 | handleException(throw_on_error); |
53 | return false; |
54 | } |
55 | } |
56 | |
57 | return true; |
58 | } |
59 | |
60 | |
61 | bool EmbeddedDictionaries::reloadImpl(const bool throw_on_error, const bool force_reload) |
62 | { |
63 | std::unique_lock lock(mutex); |
64 | |
65 | /** If you can not update the directories, then despite this, do not throw an exception (use the old directories). |
66 | * If there are no old correct directories, then when using functions that depend on them, |
67 | * will throw an exception. |
68 | * An attempt is made to load each directory separately. |
69 | */ |
70 | |
71 | LOG_INFO(log, "Loading dictionaries." ); |
72 | |
73 | bool was_exception = false; |
74 | |
75 | DictionaryReloader<RegionsHierarchies> reload_regions_hierarchies = [=, this] (const Poco::Util::AbstractConfiguration & config) |
76 | { |
77 | return geo_dictionaries_loader->reloadRegionsHierarchies(config); |
78 | }; |
79 | |
80 | if (!reloadDictionary<RegionsHierarchies>(regions_hierarchies, std::move(reload_regions_hierarchies), throw_on_error, force_reload)) |
81 | was_exception = true; |
82 | |
83 | DictionaryReloader<RegionsNames> reload_regions_names = [=, this] (const Poco::Util::AbstractConfiguration & config) |
84 | { |
85 | return geo_dictionaries_loader->reloadRegionsNames(config); |
86 | }; |
87 | |
88 | if (!reloadDictionary<RegionsNames>(regions_names, std::move(reload_regions_names), throw_on_error, force_reload)) |
89 | was_exception = true; |
90 | |
91 | if (!was_exception) |
92 | LOG_INFO(log, "Loaded dictionaries." ); |
93 | |
94 | return !was_exception; |
95 | } |
96 | |
97 | |
98 | void EmbeddedDictionaries::reloadPeriodically() |
99 | { |
100 | setThreadName("DictReload" ); |
101 | |
102 | while (true) |
103 | { |
104 | if (destroy.tryWait(cur_reload_period * 1000)) |
105 | return; |
106 | |
107 | if (reloadImpl(false)) |
108 | { |
109 | /// Success |
110 | cur_reload_period = reload_period; |
111 | is_fast_start_stage = false; |
112 | } |
113 | |
114 | if (is_fast_start_stage) |
115 | { |
116 | cur_reload_period = std::min(reload_period, 2 * cur_reload_period); /// exponentially increase delay |
117 | is_fast_start_stage = cur_reload_period < reload_period; /// leave fast start state |
118 | } |
119 | } |
120 | } |
121 | |
122 | |
123 | EmbeddedDictionaries::EmbeddedDictionaries( |
124 | std::unique_ptr<GeoDictionariesLoader> geo_dictionaries_loader_, |
125 | Context & context_, |
126 | const bool throw_on_error) |
127 | : log(&Logger::get("EmbeddedDictionaries" )) |
128 | , context(context_) |
129 | , geo_dictionaries_loader(std::move(geo_dictionaries_loader_)) |
130 | , reload_period(context_.getConfigRef().getInt("builtin_dictionaries_reload_interval" , 3600)) |
131 | { |
132 | reloadImpl(throw_on_error); |
133 | reloading_thread = ThreadFromGlobalPool([this] { reloadPeriodically(); }); |
134 | } |
135 | |
136 | |
137 | EmbeddedDictionaries::~EmbeddedDictionaries() |
138 | { |
139 | destroy.set(); |
140 | reloading_thread.join(); |
141 | } |
142 | |
143 | void EmbeddedDictionaries::reload() |
144 | { |
145 | if (!reloadImpl(true, true)) |
146 | throw Exception("Some embedded dictionaries were not successfully reloaded" , ErrorCodes::UNFINISHED); |
147 | } |
148 | |
149 | |
150 | } |
151 | |