mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-12-14 10:22:10 +00:00
1541 lines
56 KiB
C++
1541 lines
56 KiB
C++
#include "ExternalLoader.h"
|
|
|
|
#include <mutex>
|
|
#include <pcg_random.hpp>
|
|
#include <Common/Config/AbstractConfigurationComparison.h>
|
|
#include <Common/Exception.h>
|
|
#include <Common/StringUtils/StringUtils.h>
|
|
#include <Common/ThreadPool.h>
|
|
#include <Common/randomSeed.h>
|
|
#include <Common/setThreadName.h>
|
|
#include <Common/StatusInfo.h>
|
|
#include <base/chrono_io.h>
|
|
#include <Common/scope_guard_safe.h>
|
|
#include <boost/range/adaptor/map.hpp>
|
|
#include <boost/range/algorithm/copy.hpp>
|
|
#include <unordered_set>
|
|
|
|
|
|
namespace CurrentStatusInfo
|
|
{
|
|
extern const Status DictionaryStatus;
|
|
}
|
|
|
|
|
|
namespace DB
|
|
{
|
|
namespace ErrorCodes
|
|
{
|
|
extern const int LOGICAL_ERROR;
|
|
extern const int BAD_ARGUMENTS;
|
|
extern const int DICTIONARIES_WAS_NOT_LOADED;
|
|
}
|
|
|
|
|
|
namespace
|
|
{
|
|
template <typename ReturnType>
|
|
ReturnType convertTo(ExternalLoader::LoadResult result)
|
|
{
|
|
if constexpr (std::is_same_v<ReturnType, ExternalLoader::LoadResult>)
|
|
return result;
|
|
else
|
|
{
|
|
static_assert(std::is_same_v<ReturnType, ExternalLoader::LoadablePtr>);
|
|
return std::move(result.object);
|
|
}
|
|
}
|
|
|
|
template <typename ReturnType>
|
|
ReturnType convertTo(ExternalLoader::LoadResults results)
|
|
{
|
|
if constexpr (std::is_same_v<ReturnType, ExternalLoader::LoadResults>)
|
|
return results;
|
|
else
|
|
{
|
|
static_assert(std::is_same_v<ReturnType, ExternalLoader::Loadables>);
|
|
ExternalLoader::Loadables objects;
|
|
objects.reserve(results.size());
|
|
for (auto && result : results)
|
|
{
|
|
if (auto object = std::move(result.object))
|
|
objects.push_back(std::move(object));
|
|
}
|
|
return objects;
|
|
}
|
|
}
|
|
|
|
template <typename ReturnType>
|
|
ReturnType notExists(const String & name)
|
|
{
|
|
if constexpr (std::is_same_v<ReturnType, ExternalLoader::LoadResult>)
|
|
{
|
|
ExternalLoader::LoadResult res;
|
|
res.name = name;
|
|
return res;
|
|
}
|
|
else
|
|
{
|
|
static_assert(std::is_same_v<ReturnType, ExternalLoader::LoadablePtr>);
|
|
return nullptr;
|
|
}
|
|
}
|
|
|
|
|
|
/// Lock mutex only in async mode
|
|
/// In other case does nothing
|
|
struct LoadingGuardForAsyncLoad
|
|
{
|
|
std::unique_lock<std::mutex> lock;
|
|
LoadingGuardForAsyncLoad(bool async, std::mutex & mutex)
|
|
{
|
|
if (async)
|
|
lock = std::unique_lock(mutex);
|
|
}
|
|
};
|
|
}
|
|
|
|
|
|
/** Reads configurations from configuration repository and parses it.
|
|
*/
|
|
class ExternalLoader::LoadablesConfigReader : private boost::noncopyable
|
|
{
|
|
public:
|
|
LoadablesConfigReader(const String & type_name_, Poco::Logger * log_)
|
|
: type_name(type_name_), log(log_)
|
|
{
|
|
}
|
|
~LoadablesConfigReader() = default;
|
|
|
|
using Repository = IExternalLoaderConfigRepository;
|
|
|
|
void addConfigRepository(std::unique_ptr<Repository> repository)
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
auto * ptr = repository.get();
|
|
repositories.emplace(ptr, RepositoryInfo{std::move(repository), {}});
|
|
need_collect_object_configs = true;
|
|
}
|
|
|
|
void removeConfigRepository(Repository * repository)
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
auto it = repositories.find(repository);
|
|
if (it == repositories.end())
|
|
return;
|
|
repositories.erase(it);
|
|
need_collect_object_configs = true;
|
|
}
|
|
|
|
void setConfigSettings(const ExternalLoaderConfigSettings & settings_)
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
settings = settings_;
|
|
}
|
|
|
|
struct ObjectConfigs
|
|
{
|
|
std::unordered_map<String /* object's name */, std::shared_ptr<const ObjectConfig>> configs_by_name;
|
|
size_t counter = 0;
|
|
};
|
|
|
|
using ObjectConfigsPtr = std::shared_ptr<const ObjectConfigs>;
|
|
|
|
/// Reads all repositories.
|
|
ObjectConfigsPtr read()
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
readRepositories();
|
|
collectObjectConfigs();
|
|
return object_configs;
|
|
}
|
|
|
|
/// Reads only a specified repository.
|
|
/// This functions checks only a specified repository but returns configs from all repositories.
|
|
ObjectConfigsPtr read(const String & repository_name)
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
readRepositories(repository_name);
|
|
collectObjectConfigs();
|
|
return object_configs;
|
|
}
|
|
|
|
/// Reads only a specified path from a specified repository.
|
|
/// This functions checks only a specified repository but returns configs from all repositories.
|
|
ObjectConfigsPtr read(const String & repository_name, const String & path)
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
readRepositories(repository_name, path);
|
|
collectObjectConfigs();
|
|
return object_configs;
|
|
}
|
|
|
|
private:
|
|
struct FileInfo
|
|
{
|
|
Poco::Timestamp last_update_time = 0;
|
|
bool in_use = true; // Whether the `FileInfo` should be destroyed because the correspondent file is deleted.
|
|
Poco::AutoPtr<Poco::Util::AbstractConfiguration> file_contents; // Parsed contents of the file.
|
|
std::unordered_map<String /* object name */, String /* key in file_contents */> objects;
|
|
};
|
|
|
|
struct RepositoryInfo
|
|
{
|
|
std::unique_ptr<Repository> repository;
|
|
std::unordered_map<String /* path */, FileInfo> files;
|
|
};
|
|
|
|
/// Reads the repositories.
|
|
/// Checks last modification times of files and read those files which are new or changed.
|
|
void readRepositories(const std::optional<String> & only_repository_name = {}, const std::optional<String> & only_path = {})
|
|
{
|
|
for (auto & [repository, repository_info] : repositories)
|
|
{
|
|
if (only_repository_name && (repository->getName() != *only_repository_name))
|
|
continue;
|
|
|
|
for (auto & file_info : repository_info.files | boost::adaptors::map_values)
|
|
file_info.in_use = false;
|
|
|
|
Strings existing_paths;
|
|
if (only_path)
|
|
{
|
|
if (repository->exists(*only_path))
|
|
existing_paths.push_back(*only_path);
|
|
}
|
|
else
|
|
boost::copy(repository->getAllLoadablesDefinitionNames(), std::back_inserter(existing_paths));
|
|
|
|
for (const auto & path : existing_paths)
|
|
{
|
|
auto it = repository_info.files.find(path);
|
|
if (it != repository_info.files.end())
|
|
{
|
|
FileInfo & file_info = it->second;
|
|
if (readFileInfo(file_info, *repository, path))
|
|
need_collect_object_configs = true;
|
|
}
|
|
else
|
|
{
|
|
FileInfo file_info;
|
|
if (readFileInfo(file_info, *repository, path))
|
|
{
|
|
repository_info.files.emplace(path, std::move(file_info));
|
|
need_collect_object_configs = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
Strings deleted_paths;
|
|
for (auto & [path, file_info] : repository_info.files)
|
|
{
|
|
if (file_info.in_use)
|
|
continue;
|
|
|
|
if (only_path && (*only_path != path))
|
|
continue;
|
|
|
|
deleted_paths.emplace_back(path);
|
|
}
|
|
|
|
if (!deleted_paths.empty())
|
|
{
|
|
for (const String & deleted_path : deleted_paths)
|
|
repository_info.files.erase(deleted_path);
|
|
need_collect_object_configs = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Reads a file, returns true if the file is new or changed.
|
|
bool readFileInfo(
|
|
FileInfo & file_info,
|
|
IExternalLoaderConfigRepository & repository,
|
|
const String & path) const
|
|
{
|
|
try
|
|
{
|
|
if (path.empty() || !repository.exists(path))
|
|
{
|
|
LOG_WARNING(log, "Config file '{}' does not exist", path);
|
|
return false;
|
|
}
|
|
|
|
auto update_time_from_repository = repository.getUpdateTime(path);
|
|
|
|
// We can't count on that the mtime increases or that it has
|
|
// a particular relation to system time, so just check for strict
|
|
// equality.
|
|
// Note that on 1.x versions on Poco, the granularity of update
|
|
// time is one second, so the window where we can miss the changes
|
|
// is that wide (i.e. when we read the file and after that it
|
|
// is updated, but in the same second).
|
|
// The solution to this is probably switching to std::filesystem
|
|
// -- the work is underway to do so.
|
|
if (update_time_from_repository == file_info.last_update_time)
|
|
{
|
|
file_info.in_use = true;
|
|
return false;
|
|
}
|
|
|
|
LOG_TRACE(log, "Loading config file '{}'.", path);
|
|
file_info.file_contents = repository.load(path);
|
|
auto & file_contents = *file_info.file_contents;
|
|
|
|
/// get all objects' definitions
|
|
Poco::Util::AbstractConfiguration::Keys keys;
|
|
file_contents.keys(keys);
|
|
|
|
/// for each object defined in repositories
|
|
std::unordered_map<String, String> objects;
|
|
for (const auto & key : keys)
|
|
{
|
|
if (!startsWith(key, settings.external_config))
|
|
{
|
|
if (!startsWith(key, "comment") && !startsWith(key, "include_from"))
|
|
LOG_WARNING(log, "{}: file contains unknown node '{}', expected '{}'", path, key, settings.external_config);
|
|
continue;
|
|
}
|
|
|
|
/// Use uuid as name if possible
|
|
String object_uuid = file_contents.getString(key + "." + settings.external_uuid, "");
|
|
String object_name;
|
|
if (object_uuid.empty())
|
|
object_name = file_contents.getString(key + "." + settings.external_name);
|
|
else
|
|
object_name = object_uuid;
|
|
if (object_name.empty())
|
|
{
|
|
LOG_WARNING(log, "{}: node '{}' defines {} with an empty name. It's not allowed", path, key, type_name);
|
|
continue;
|
|
}
|
|
|
|
if (object_uuid.empty())
|
|
{
|
|
String database;
|
|
if (!settings.external_database.empty())
|
|
database = file_contents.getString(key + "." + settings.external_database, "");
|
|
if (!database.empty())
|
|
object_name = database + "." + object_name;
|
|
}
|
|
|
|
objects.emplace(object_name, key);
|
|
}
|
|
|
|
file_info.objects = std::move(objects);
|
|
file_info.last_update_time = update_time_from_repository;
|
|
file_info.in_use = true;
|
|
return true;
|
|
}
|
|
catch (...)
|
|
{
|
|
tryLogCurrentException(log, "Failed to load config file '" + path + "'");
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/// Builds a map of current configurations of objects.
|
|
void collectObjectConfigs()
|
|
{
|
|
if (!need_collect_object_configs)
|
|
return;
|
|
need_collect_object_configs = false;
|
|
|
|
// Generate new result.
|
|
auto new_configs = std::make_shared<ObjectConfigs>();
|
|
|
|
for (const auto & [repository, repository_info] : repositories)
|
|
{
|
|
for (const auto & [path, file_info] : repository_info.files)
|
|
{
|
|
for (const auto & [object_name, key_in_config] : file_info.objects)
|
|
{
|
|
auto already_added_it = new_configs->configs_by_name.find(object_name);
|
|
if (already_added_it == new_configs->configs_by_name.end())
|
|
{
|
|
auto new_config = std::make_shared<ObjectConfig>();
|
|
new_config->config = file_info.file_contents;
|
|
new_config->key_in_config = key_in_config;
|
|
new_config->repository_name = repository->getName();
|
|
new_config->from_temp_repository = repository->isTemporary();
|
|
new_config->path = path;
|
|
new_configs->configs_by_name.emplace(object_name, std::move(new_config));
|
|
}
|
|
else
|
|
{
|
|
const auto & already_added = already_added_it->second;
|
|
if (!already_added->from_temp_repository && !repository->isTemporary())
|
|
{
|
|
if (path == already_added->path && repository->getName() == already_added->repository_name)
|
|
LOG_WARNING(log, "{} '{}' is found twice in the same file '{}'",
|
|
type_name, object_name, path);
|
|
else
|
|
LOG_WARNING(log, "{} '{}' is found both in file '{}' and '{}'",
|
|
type_name, object_name, already_added->path, path);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
new_configs->counter = counter++;
|
|
object_configs = new_configs;
|
|
}
|
|
|
|
const String type_name;
|
|
Poco::Logger * log;
|
|
|
|
std::mutex mutex;
|
|
ExternalLoaderConfigSettings settings;
|
|
std::unordered_map<Repository *, RepositoryInfo> repositories;
|
|
ObjectConfigsPtr object_configs;
|
|
bool need_collect_object_configs = false;
|
|
size_t counter = 0;
|
|
};
|
|
|
|
|
|
/** Manages loading and reloading objects. Uses configurations from the class LoadablesConfigReader.
|
|
* Supports parallel loading.
|
|
*/
|
|
class ExternalLoader::LoadingDispatcher : private boost::noncopyable
|
|
{
|
|
public:
|
|
/// Called to load or reload an object.
|
|
using CreateObjectFunction = std::function<LoadablePtr(
|
|
const String & /* name */, const ObjectConfig & /* config */, const LoadablePtr & /* previous_version */)>;
|
|
|
|
LoadingDispatcher(
|
|
const CreateObjectFunction & create_object_function_,
|
|
const String & type_name_,
|
|
Poco::Logger * log_)
|
|
: create_object(create_object_function_)
|
|
, type_name(type_name_)
|
|
, log(log_)
|
|
{
|
|
}
|
|
|
|
~LoadingDispatcher()
|
|
{
|
|
std::unique_lock lock{mutex};
|
|
infos.clear(); /// We clear this map to tell the threads that we don't want any load results anymore.
|
|
|
|
/// Wait for all the threads to finish.
|
|
while (!loading_threads.empty())
|
|
{
|
|
auto it = loading_threads.begin();
|
|
auto thread = std::move(it->second);
|
|
loading_threads.erase(it);
|
|
lock.unlock();
|
|
event.notify_all();
|
|
thread.join();
|
|
lock.lock();
|
|
}
|
|
}
|
|
|
|
using ObjectConfigsPtr = LoadablesConfigReader::ObjectConfigsPtr;
|
|
|
|
/// Sets new configurations for all the objects.
|
|
void setConfiguration(const ObjectConfigsPtr & new_configs)
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
if (configs == new_configs)
|
|
return;
|
|
|
|
/// The following check prevents a race when two threads are trying to update configuration
|
|
/// at almost the same time:
|
|
/// 1) first thread reads a configuration (for example as a part of periodic updates)
|
|
/// 2) second thread sets a new configuration (for example after executing CREATE DICTIONARY)
|
|
/// 3) first thread sets the configuration it read in 1) and thus discards the changes made in 2).
|
|
/// So we use `counter` here to ensure we exchange the current configuration only for a newer one.
|
|
if (configs && (configs->counter >= new_configs->counter))
|
|
return;
|
|
|
|
configs = new_configs;
|
|
|
|
std::vector<String> removed_names;
|
|
for (auto & [name, info] : infos)
|
|
{
|
|
auto new_config_it = new_configs->configs_by_name.find(name);
|
|
if (new_config_it == new_configs->configs_by_name.end())
|
|
{
|
|
removed_names.emplace_back(name);
|
|
}
|
|
else
|
|
{
|
|
const auto & new_config = new_config_it->second;
|
|
bool config_is_same = isSameConfiguration(*info.config->config, info.config->key_in_config, *new_config->config, new_config->key_in_config);
|
|
info.config = new_config;
|
|
if (!config_is_same)
|
|
{
|
|
if (info.triedToLoad())
|
|
{
|
|
/// The object has been tried to load before, so it is currently in use or was in use
|
|
/// and we should try to reload it with the new config.
|
|
LOG_TRACE(log, "Will reload '{}' because its configuration has been changed and there were attempts to load it before", name);
|
|
startLoading(info, true);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Insert to the map those objects which added to the new configuration.
|
|
for (const auto & [name, config] : new_configs->configs_by_name)
|
|
{
|
|
if (infos.find(name) == infos.end())
|
|
{
|
|
Info & info = infos.emplace(name, Info{name, config}).first->second;
|
|
if (always_load_everything)
|
|
{
|
|
LOG_TRACE(log, "Will load '{}' because always_load_everything flag is set.", name);
|
|
startLoading(info);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Remove from the map those objects which were removed from the configuration.
|
|
for (const String & name : removed_names)
|
|
{
|
|
if (auto it = infos.find(name); it != infos.end())
|
|
{
|
|
const auto & info = it->second;
|
|
if (info.loaded() || info.isLoading())
|
|
LOG_TRACE(log, "Unloading '{}' because its configuration has been removed or detached", name);
|
|
infos.erase(it);
|
|
}
|
|
}
|
|
|
|
/// Maybe we have just added new objects which require to be loaded
|
|
/// or maybe we have just removed object which were been loaded,
|
|
/// so we should notify `event` to recheck conditions in load() and loadAll() now.
|
|
event.notify_all();
|
|
}
|
|
|
|
/// Sets whether all the objects from the configuration should be always loaded (even if they aren't used).
|
|
void enableAlwaysLoadEverything(bool enable)
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
if (always_load_everything == enable)
|
|
return;
|
|
|
|
always_load_everything = enable;
|
|
|
|
if (enable)
|
|
{
|
|
/// Start loading all the objects which were not loaded yet.
|
|
for (auto & [name, info] : infos)
|
|
if (!info.triedToLoad())
|
|
startLoading(info);
|
|
}
|
|
}
|
|
|
|
/// Sets whether the objects should be loaded asynchronously, each loading in a new thread (from the thread pool).
|
|
void enableAsyncLoading(bool enable)
|
|
{
|
|
enable_async_loading = enable;
|
|
}
|
|
|
|
/// Returns the status of the object.
|
|
/// If the object has not been loaded yet then the function returns Status::NOT_LOADED.
|
|
/// If the specified name isn't found in the configuration then the function returns Status::NOT_EXIST.
|
|
Status getCurrentStatus(const String & name) const
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
const Info * info = getInfo(name);
|
|
if (!info)
|
|
return Status::NOT_EXIST;
|
|
return info->status();
|
|
}
|
|
|
|
/// Returns the load result of the object.
|
|
template <typename ReturnType>
|
|
ReturnType getLoadResult(const String & name) const
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
const Info * info = getInfo(name);
|
|
if (!info)
|
|
return notExists<ReturnType>(name);
|
|
return info->getLoadResult<ReturnType>();
|
|
}
|
|
|
|
/// Returns all the load results as a map.
|
|
/// The function doesn't load anything, it just returns the current load results as is.
|
|
template <typename ReturnType>
|
|
ReturnType getLoadResults(const FilterByNameFunction & filter) const
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
return collectLoadResults<ReturnType>(filter);
|
|
}
|
|
|
|
size_t getNumberOfLoadedObjects() const
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
size_t count = 0;
|
|
for (const auto & name_and_info : infos)
|
|
{
|
|
const auto & info = name_and_info.second;
|
|
if (info.loaded())
|
|
++count;
|
|
}
|
|
return count;
|
|
}
|
|
|
|
bool hasLoadedObjects() const
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
for (const auto & name_info : infos)
|
|
if (name_info.second.loaded())
|
|
return true;
|
|
return false;
|
|
}
|
|
|
|
Strings getAllTriedToLoadNames() const
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
Strings names;
|
|
for (const auto & [name, info] : infos)
|
|
if (info.triedToLoad())
|
|
names.push_back(name);
|
|
return names;
|
|
}
|
|
|
|
size_t getNumberOfObjects() const
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
return infos.size();
|
|
}
|
|
|
|
/// Tries to load a specified object during the timeout.
|
|
template <typename ReturnType>
|
|
ReturnType tryLoad(const String & name, Duration timeout)
|
|
{
|
|
std::unique_lock lock{mutex};
|
|
Info * info = loadImpl(name, timeout, false, lock);
|
|
if (!info)
|
|
return notExists<ReturnType>(name);
|
|
return info->getLoadResult<ReturnType>();
|
|
}
|
|
|
|
template <typename ReturnType>
|
|
ReturnType tryLoad(const FilterByNameFunction & filter, Duration timeout)
|
|
{
|
|
std::unique_lock lock{mutex};
|
|
loadImpl(filter, timeout, false, lock);
|
|
return collectLoadResults<ReturnType>(filter);
|
|
}
|
|
|
|
/// Tries to load or reload a specified object.
|
|
template <typename ReturnType>
|
|
ReturnType tryLoadOrReload(const String & name, Duration timeout)
|
|
{
|
|
std::unique_lock lock{mutex};
|
|
Info * info = loadImpl(name, timeout, true, lock);
|
|
if (!info)
|
|
return notExists<ReturnType>(name);
|
|
return info->getLoadResult<ReturnType>();
|
|
}
|
|
|
|
template <typename ReturnType>
|
|
ReturnType tryLoadOrReload(const FilterByNameFunction & filter, Duration timeout)
|
|
{
|
|
std::unique_lock lock{mutex};
|
|
loadImpl(filter, timeout, true, lock);
|
|
return collectLoadResults<ReturnType>(filter);
|
|
}
|
|
|
|
bool has(const String & name) const
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
return infos.contains(name);
|
|
}
|
|
|
|
/// Starts reloading all the object which update time is earlier than now.
|
|
/// The function doesn't touch the objects which were never tried to load.
|
|
void reloadOutdated()
|
|
{
|
|
/// Iterate through all the objects and find loaded ones which should be checked if they need update.
|
|
std::unordered_map<LoadablePtr, bool> should_update_map;
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
TimePoint now = std::chrono::system_clock::now();
|
|
for (const auto & name_and_info : infos)
|
|
{
|
|
const auto & info = name_and_info.second;
|
|
if ((now >= info.next_update_time) && !info.isLoading() && info.loaded())
|
|
should_update_map.emplace(info.object, info.failedToReload());
|
|
}
|
|
}
|
|
|
|
/// Find out which of the loaded objects were modified.
|
|
/// We couldn't perform these checks while we were building `should_update_map` because
|
|
/// the `mutex` should be unlocked while we're calling the function object->isModified()
|
|
for (auto & [object, should_update_flag] : should_update_map)
|
|
{
|
|
try
|
|
{
|
|
/// Maybe already true, if we have an exception
|
|
if (!should_update_flag)
|
|
should_update_flag = object->isModified();
|
|
}
|
|
catch (...)
|
|
{
|
|
tryLogCurrentException(log, "Could not check if " + type_name + " '" + object->getLoadableName() + "' was modified");
|
|
/// Cannot check isModified, so update
|
|
should_update_flag = true;
|
|
}
|
|
}
|
|
|
|
/// Iterate through all the objects again and either start loading or just set `next_update_time`.
|
|
{
|
|
std::lock_guard lock{mutex};
|
|
TimePoint now = std::chrono::system_clock::now();
|
|
for (auto & [name, info] : infos)
|
|
{
|
|
if ((now >= info.next_update_time) && !info.isLoading())
|
|
{
|
|
if (info.loaded())
|
|
{
|
|
auto it = should_update_map.find(info.object);
|
|
if (it == should_update_map.end())
|
|
continue; /// Object has been just loaded (it wasn't loaded while we were building the map `should_update_map`), so we don't have to reload it right now.
|
|
|
|
bool should_update_flag = it->second;
|
|
if (!should_update_flag)
|
|
{
|
|
info.next_update_time = calculateNextUpdateTime(info.object, info.error_count);
|
|
LOG_TRACE(log, "Object '{}' not modified, will not reload. Next update at {}", info.name, to_string(info.next_update_time));
|
|
continue;
|
|
}
|
|
|
|
/// Object was modified or it was failed to reload last time, so it should be reloaded.
|
|
startLoading(info);
|
|
}
|
|
else if (info.failed())
|
|
{
|
|
/// Object was never loaded successfully and should be reloaded.
|
|
startLoading(info);
|
|
}
|
|
LOG_TRACE(log, "Object '{}' is neither loaded nor failed, so it will not be reloaded as outdated.", info.name);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private:
|
|
struct Info
|
|
{
|
|
Info(const String & name_, const std::shared_ptr<const ObjectConfig> & config_) : name(name_), config(config_) {}
|
|
|
|
bool loaded() const { return object != nullptr; }
|
|
bool failed() const { return !object && exception; }
|
|
bool loadedOrFailed() const { return loaded() || failed(); }
|
|
bool triedToLoad() const { return loaded() || failed() || isLoading(); }
|
|
bool failedToReload() const { return loaded() && exception != nullptr; }
|
|
bool isLoading() const { return loading_id > state_id; }
|
|
|
|
Status status() const
|
|
{
|
|
if (object)
|
|
return isLoading() ? Status::LOADED_AND_RELOADING : Status::LOADED;
|
|
else if (exception)
|
|
return isLoading() ? Status::FAILED_AND_RELOADING : Status::FAILED;
|
|
else
|
|
return isLoading() ? Status::LOADING : Status::NOT_LOADED;
|
|
}
|
|
|
|
Duration loadingDuration() const
|
|
{
|
|
if (isLoading())
|
|
return std::chrono::duration_cast<Duration>(std::chrono::system_clock::now() - loading_start_time);
|
|
return std::chrono::duration_cast<Duration>(loading_end_time - loading_start_time);
|
|
}
|
|
|
|
template <typename ReturnType>
|
|
ReturnType getLoadResult() const
|
|
{
|
|
if constexpr (std::is_same_v<ReturnType, LoadResult>)
|
|
{
|
|
LoadResult result;
|
|
result.name = name;
|
|
result.status = status();
|
|
result.object = object;
|
|
result.exception = exception;
|
|
result.loading_start_time = loading_start_time;
|
|
result.last_successful_update_time = last_successful_update_time;
|
|
result.loading_duration = loadingDuration();
|
|
result.config = config;
|
|
return result;
|
|
}
|
|
else
|
|
{
|
|
static_assert(std::is_same_v<ReturnType, ExternalLoader::LoadablePtr>);
|
|
return object;
|
|
}
|
|
}
|
|
|
|
String name;
|
|
LoadablePtr object;
|
|
std::shared_ptr<const ObjectConfig> config;
|
|
TimePoint loading_start_time;
|
|
TimePoint loading_end_time;
|
|
TimePoint last_successful_update_time;
|
|
size_t state_id = 0; /// Index of the current state of this `info`, this index is incremented every loading.
|
|
size_t loading_id = 0; /// The value which will be stored in `state_id` after finishing the current loading.
|
|
size_t error_count = 0; /// Numbers of errors since last successful loading.
|
|
std::exception_ptr exception; /// Last error occurred.
|
|
TimePoint next_update_time = TimePoint::max(); /// Time of the next update, `TimePoint::max()` means "never".
|
|
};
|
|
|
|
Info * getInfo(const String & name)
|
|
{
|
|
auto it = infos.find(name);
|
|
if (it == infos.end())
|
|
return nullptr;
|
|
return &it->second;
|
|
}
|
|
|
|
const Info * getInfo(const String & name) const
|
|
{
|
|
auto it = infos.find(name);
|
|
if (it == infos.end())
|
|
return nullptr;
|
|
return &it->second;
|
|
}
|
|
|
|
template <typename ReturnType>
|
|
ReturnType collectLoadResults(const FilterByNameFunction & filter) const
|
|
{
|
|
ReturnType results;
|
|
results.reserve(infos.size());
|
|
for (const auto & [name, info] : infos)
|
|
{
|
|
if (!filter || filter(name))
|
|
{
|
|
auto result = info.template getLoadResult<typename ReturnType::value_type>();
|
|
if constexpr (std::is_same_v<typename ReturnType::value_type, LoadablePtr>)
|
|
{
|
|
if (!result)
|
|
continue;
|
|
}
|
|
results.emplace_back(std::move(result));
|
|
}
|
|
}
|
|
return results;
|
|
}
|
|
|
|
Info * loadImpl(const String & name, Duration timeout, bool forced_to_reload, std::unique_lock<std::mutex> & lock)
|
|
{
|
|
std::optional<size_t> min_id;
|
|
Info * info = nullptr;
|
|
auto pred = [&]
|
|
{
|
|
info = getInfo(name);
|
|
if (!info)
|
|
return true; /// stop
|
|
|
|
if (!min_id)
|
|
min_id = getMinIDToFinishLoading(forced_to_reload);
|
|
|
|
if (info->loading_id < min_id)
|
|
startLoading(*info, forced_to_reload, *min_id);
|
|
|
|
/// Wait for the next event if loading wasn't completed, or stop otherwise.
|
|
return (info->state_id >= min_id);
|
|
};
|
|
|
|
if (timeout == WAIT)
|
|
event.wait(lock, pred);
|
|
else
|
|
event.wait_for(lock, timeout, pred);
|
|
|
|
return info;
|
|
}
|
|
|
|
void loadImpl(const FilterByNameFunction & filter, Duration timeout, bool forced_to_reload, std::unique_lock<std::mutex> & lock)
|
|
{
|
|
std::optional<size_t> min_id;
|
|
auto pred = [&]
|
|
{
|
|
if (!min_id)
|
|
min_id = getMinIDToFinishLoading(forced_to_reload);
|
|
|
|
bool all_ready = true;
|
|
for (auto & [name, info] : infos)
|
|
{
|
|
if (filter && !filter(name))
|
|
continue;
|
|
|
|
if (info.loading_id < min_id)
|
|
startLoading(info, forced_to_reload, *min_id);
|
|
|
|
all_ready &= (info.state_id >= min_id);
|
|
}
|
|
return all_ready;
|
|
};
|
|
|
|
if (timeout == WAIT)
|
|
event.wait(lock, pred);
|
|
else
|
|
event.wait_for(lock, timeout, pred);
|
|
}
|
|
|
|
/// When state_id >= getMinIDToFinishLoading() the loading is considered as finished.
|
|
size_t getMinIDToFinishLoading(bool forced_to_reload) const
|
|
{
|
|
if (forced_to_reload)
|
|
{
|
|
/// We need to force reloading, that's why we return next_id_counter here
|
|
/// (because info.state_id < next_id_counter for any info).
|
|
return next_id_counter;
|
|
}
|
|
|
|
/// The loading of an object can cause the loading of another object.
|
|
/// We use the same "min_id" in this case to allows reloading multiple objects at once
|
|
/// taking into account their dependencies.
|
|
auto it = min_id_to_finish_loading_dependencies.find(std::this_thread::get_id());
|
|
if (it != min_id_to_finish_loading_dependencies.end())
|
|
return it->second;
|
|
|
|
/// We just need the first loading to be finished, that's why we return 1 here
|
|
/// (because info.state_id >= 1 since the first loading is finished, successfully or not).
|
|
return 1;
|
|
}
|
|
|
|
void startLoading(Info & info, bool forced_to_reload = false, size_t min_id_to_finish_loading_dependencies_ = 1)
|
|
{
|
|
if (info.isLoading())
|
|
{
|
|
LOG_TRACE(log, "The object '{}' is already being loaded, force = {}.", info.name, forced_to_reload);
|
|
|
|
if (!forced_to_reload)
|
|
{
|
|
return;
|
|
}
|
|
|
|
cancelLoading(info);
|
|
}
|
|
|
|
putBackFinishedThreadsToPool();
|
|
|
|
/// All loadings have unique loading IDs.
|
|
size_t loading_id = next_id_counter++;
|
|
info.loading_id = loading_id;
|
|
info.loading_start_time = std::chrono::system_clock::now();
|
|
info.loading_end_time = TimePoint{};
|
|
|
|
LOG_TRACE(log, "Will load the object '{}' {}, force = {}, loading_id = {}", info.name, (enable_async_loading ? std::string("in background") : "immediately"), forced_to_reload, info.loading_id);
|
|
|
|
if (enable_async_loading)
|
|
{
|
|
/// Put a job to the thread pool for the loading.
|
|
auto thread = ThreadFromGlobalPool{&LoadingDispatcher::doLoading, this, info.name, loading_id, forced_to_reload, min_id_to_finish_loading_dependencies_, true, CurrentThread::getGroup()};
|
|
loading_threads.try_emplace(loading_id, std::move(thread));
|
|
}
|
|
else
|
|
{
|
|
/// Perform the loading immediately.
|
|
doLoading(info.name, loading_id, forced_to_reload, min_id_to_finish_loading_dependencies_, false);
|
|
}
|
|
}
|
|
|
|
void putBackFinishedThreadsToPool()
|
|
{
|
|
for (auto loading_id : recently_finished_loadings)
|
|
{
|
|
auto it = loading_threads.find(loading_id);
|
|
if (it != loading_threads.end())
|
|
{
|
|
auto thread = std::move(it->second);
|
|
loading_threads.erase(it);
|
|
thread.join(); /// It's very likely that `thread` has already finished.
|
|
}
|
|
}
|
|
recently_finished_loadings.clear();
|
|
}
|
|
|
|
static void cancelLoading(Info & info)
|
|
{
|
|
if (!info.isLoading())
|
|
return;
|
|
|
|
/// In fact we cannot actually CANCEL the loading (because it's possibly already being performed in another thread).
|
|
/// But we can reset the `loading_id` and doLoading() will understand it as a signal to stop loading.
|
|
info.loading_id = info.state_id;
|
|
info.loading_end_time = std::chrono::system_clock::now();
|
|
}
|
|
|
|
/// Does the loading, possibly in the separate thread.
|
|
void doLoading(const String & name, size_t loading_id, bool forced_to_reload, size_t min_id_to_finish_loading_dependencies_, bool async, ThreadGroupStatusPtr thread_group = {})
|
|
{
|
|
SCOPE_EXIT_SAFE(
|
|
if (thread_group)
|
|
CurrentThread::detachQueryIfNotDetached();
|
|
);
|
|
|
|
if (thread_group)
|
|
CurrentThread::attachTo(thread_group);
|
|
|
|
LOG_TRACE(log, "Start loading object '{}'", name);
|
|
try
|
|
{
|
|
/// Prepare for loading.
|
|
std::optional<Info> info;
|
|
{
|
|
LoadingGuardForAsyncLoad lock(async, mutex);
|
|
info = prepareToLoadSingleObject(name, loading_id, min_id_to_finish_loading_dependencies_, lock);
|
|
if (!info)
|
|
{
|
|
LOG_TRACE(log, "Could not lock object '{}' for loading", name);
|
|
return;
|
|
}
|
|
}
|
|
|
|
/// Previous version can be used as the base for new loading, enabling loading only part of data.
|
|
auto previous_version_as_base_for_loading = info->object;
|
|
if (forced_to_reload)
|
|
previous_version_as_base_for_loading = nullptr; /// Need complete reloading, cannot use the previous version.
|
|
|
|
/// Loading.
|
|
auto [new_object, new_exception] = loadSingleObject(name, *info->config, previous_version_as_base_for_loading);
|
|
if (!new_object && !new_exception)
|
|
throw Exception("No object created and no exception raised for " + type_name, ErrorCodes::LOGICAL_ERROR);
|
|
|
|
/// Saving the result of the loading.
|
|
{
|
|
LoadingGuardForAsyncLoad lock(async, mutex);
|
|
saveResultOfLoadingSingleObject(name, loading_id, info->object, new_object, new_exception, info->error_count, lock);
|
|
finishLoadingSingleObject(name, loading_id, lock);
|
|
}
|
|
event.notify_all();
|
|
}
|
|
catch (...)
|
|
{
|
|
LoadingGuardForAsyncLoad lock(async, mutex);
|
|
finishLoadingSingleObject(name, loading_id, lock);
|
|
throw;
|
|
}
|
|
}
|
|
|
|
/// Returns single object info, checks loading_id and name.
|
|
std::optional<Info> prepareToLoadSingleObject(
|
|
const String & name, size_t loading_id, size_t min_id_to_finish_loading_dependencies_, const LoadingGuardForAsyncLoad &)
|
|
{
|
|
Info * info = getInfo(name);
|
|
/// We check here if this is exactly the same loading as we planned to perform.
|
|
/// This check is necessary because the object could be removed or load with another config before this thread even starts.
|
|
if (!info || !info->isLoading() || (info->loading_id != loading_id))
|
|
return {};
|
|
|
|
min_id_to_finish_loading_dependencies[std::this_thread::get_id()] = min_id_to_finish_loading_dependencies_;
|
|
return *info;
|
|
}
|
|
|
|
/// Load one object, returns object ptr or exception.
|
|
std::pair<LoadablePtr, std::exception_ptr>
|
|
loadSingleObject(const String & name, const ObjectConfig & config, LoadablePtr previous_version)
|
|
{
|
|
/// Use `create_function` to perform the actual loading.
|
|
/// It's much better to do it with `mutex` unlocked because the loading can take a lot of time
|
|
/// and require access to other objects.
|
|
LoadablePtr new_object;
|
|
std::exception_ptr new_exception;
|
|
try
|
|
{
|
|
new_object = create_object(name, config, previous_version);
|
|
}
|
|
catch (...)
|
|
{
|
|
new_exception = std::current_exception();
|
|
}
|
|
return std::make_pair(new_object, new_exception);
|
|
}
|
|
|
|
/// Saves the result of the loading, calculates the time of the next update, and handles errors.
|
|
void saveResultOfLoadingSingleObject(
|
|
const String & name,
|
|
size_t loading_id,
|
|
LoadablePtr previous_version,
|
|
LoadablePtr new_object,
|
|
std::exception_ptr new_exception,
|
|
size_t error_count,
|
|
const LoadingGuardForAsyncLoad &)
|
|
{
|
|
/// Calculate a new update time.
|
|
TimePoint next_update_time;
|
|
try
|
|
{
|
|
if (new_exception)
|
|
++error_count;
|
|
else
|
|
error_count = 0;
|
|
|
|
LoadablePtr object = previous_version;
|
|
if (new_object)
|
|
object = new_object;
|
|
|
|
next_update_time = calculateNextUpdateTime(object, error_count);
|
|
}
|
|
catch (...)
|
|
{
|
|
tryLogCurrentException(log, "Cannot find out when the " + type_name + " '" + name + "' should be updated");
|
|
next_update_time = TimePoint::max();
|
|
}
|
|
|
|
|
|
Info * info = getInfo(name);
|
|
|
|
/// We should check if this is still the same loading as we were doing.
|
|
/// This is necessary because the object could be removed or load with another config while the `mutex` was unlocked.
|
|
if (!info)
|
|
{
|
|
LOG_TRACE(log, "Next update time for '{}' will not be set because this object was not found.", name);
|
|
return;
|
|
}
|
|
if (!info->isLoading())
|
|
{
|
|
LOG_TRACE(log, "Next update time for '{}' will not be set because this object is not currently loading.", name);
|
|
return;
|
|
}
|
|
if (info->loading_id != loading_id)
|
|
{
|
|
LOG_TRACE(log, "Next update time for '{}' will not be set because this object's current loading_id {} is different from the specified {}.", name, info->loading_id, loading_id);
|
|
return;
|
|
}
|
|
|
|
if (new_exception)
|
|
{
|
|
auto next_update_time_description = [next_update_time]
|
|
{
|
|
if (next_update_time == TimePoint::max())
|
|
return String();
|
|
return ", next update is scheduled at " + to_string(next_update_time);
|
|
};
|
|
if (previous_version)
|
|
tryLogException(new_exception, log, "Could not update " + type_name + " '" + name + "'"
|
|
", leaving the previous version" + next_update_time_description());
|
|
else
|
|
tryLogException(new_exception, log, "Could not load " + type_name + " '" + name + "'" + next_update_time_description());
|
|
}
|
|
|
|
if (new_object)
|
|
info->object = new_object;
|
|
|
|
info->exception = new_exception;
|
|
info->error_count = error_count;
|
|
const auto current_time = std::chrono::system_clock::now();
|
|
info->loading_end_time = current_time;
|
|
if (!info->exception)
|
|
info->last_successful_update_time = current_time;
|
|
info->state_id = info->loading_id;
|
|
info->next_update_time = next_update_time;
|
|
LOG_TRACE(log, "Next update time for '{}' was set to {}", info->name, to_string(next_update_time));
|
|
}
|
|
|
|
/// Removes the references to the loading thread from the maps.
|
|
void finishLoadingSingleObject(const String & name, size_t loading_id, const LoadingGuardForAsyncLoad &)
|
|
{
|
|
Info * info = getInfo(name);
|
|
if (info && (info->loading_id == loading_id))
|
|
{
|
|
info->loading_id = info->state_id;
|
|
CurrentStatusInfo::set(CurrentStatusInfo::DictionaryStatus, name, static_cast<Int8>(info->status()));
|
|
}
|
|
min_id_to_finish_loading_dependencies.erase(std::this_thread::get_id());
|
|
|
|
/// Add `loading_id` to the list of recently finished loadings.
|
|
/// This list is used to later put the threads which finished loading back to the thread pool.
|
|
/// (We can't put the loading thread back to the thread pool immediately here because at this point
|
|
/// the loading thread is about to finish but it's not finished yet right now.)
|
|
recently_finished_loadings.push_back(loading_id);
|
|
}
|
|
|
|
/// Calculate next update time for loaded_object. Can be called without mutex locking,
|
|
/// because single loadable can be loaded in single thread only.
|
|
TimePoint calculateNextUpdateTime(const LoadablePtr & loaded_object, size_t error_count) const
|
|
{
|
|
static constexpr auto never = TimePoint::max();
|
|
|
|
if (loaded_object)
|
|
{
|
|
if (!loaded_object->supportUpdates())
|
|
{
|
|
LOG_TRACE(log, "Supposed update time for '{}' is never (loaded, does not support updates)", loaded_object->getLoadableName());
|
|
|
|
return never;
|
|
}
|
|
|
|
/// do not update loadable objects with zero as lifetime
|
|
const auto & lifetime = loaded_object->getLifetime();
|
|
if (lifetime.min_sec == 0 && lifetime.max_sec == 0)
|
|
{
|
|
LOG_TRACE(log, "Supposed update time for '{}' is never (loaded, lifetime 0)", loaded_object->getLoadableName());
|
|
return never;
|
|
}
|
|
|
|
if (!error_count)
|
|
{
|
|
std::uniform_int_distribution<UInt64> distribution{lifetime.min_sec, lifetime.max_sec};
|
|
auto result = std::chrono::system_clock::now() + std::chrono::seconds{distribution(rnd_engine)};
|
|
LOG_TRACE(log, "Supposed update time for '{}' is {} (loaded, lifetime [{}, {}], no errors)",
|
|
loaded_object->getLoadableName(), to_string(result), lifetime.min_sec, lifetime.max_sec);
|
|
return result;
|
|
}
|
|
|
|
auto result = std::chrono::system_clock::now() + std::chrono::seconds(calculateDurationWithBackoff(rnd_engine, error_count));
|
|
LOG_TRACE(log, "Supposed update time for '{}' is {} (backoff, {} errors)", loaded_object->getLoadableName(), to_string(result), error_count);
|
|
return result;
|
|
}
|
|
else
|
|
{
|
|
auto result = std::chrono::system_clock::now() + std::chrono::seconds(calculateDurationWithBackoff(rnd_engine, error_count));
|
|
LOG_TRACE(log, "Supposed update time for unspecified object is {} (backoff, {} errors.", to_string(result), error_count);
|
|
return result;
|
|
}
|
|
}
|
|
|
|
const CreateObjectFunction create_object;
|
|
const String type_name;
|
|
Poco::Logger * log;
|
|
|
|
mutable std::mutex mutex;
|
|
std::condition_variable event;
|
|
ObjectConfigsPtr configs;
|
|
std::unordered_map<String, Info> infos;
|
|
bool always_load_everything = false;
|
|
std::atomic<bool> enable_async_loading = false;
|
|
std::unordered_map<size_t, ThreadFromGlobalPool> loading_threads;
|
|
std::vector<size_t> recently_finished_loadings;
|
|
std::unordered_map<std::thread::id, size_t> min_id_to_finish_loading_dependencies;
|
|
size_t next_id_counter = 1; /// should always be > 0
|
|
mutable pcg64 rnd_engine{randomSeed()};
|
|
};
|
|
|
|
|
|
class ExternalLoader::PeriodicUpdater : private boost::noncopyable
|
|
{
|
|
public:
|
|
static constexpr UInt64 check_period_sec = 5;
|
|
|
|
PeriodicUpdater(LoadablesConfigReader & config_files_reader_, LoadingDispatcher & loading_dispatcher_)
|
|
: config_files_reader(config_files_reader_), loading_dispatcher(loading_dispatcher_)
|
|
{
|
|
}
|
|
|
|
~PeriodicUpdater() { enable(false); }
|
|
|
|
void enable(bool enable_)
|
|
{
|
|
std::unique_lock lock{mutex};
|
|
enabled = enable_;
|
|
|
|
if (enable_)
|
|
{
|
|
if (!thread.joinable())
|
|
{
|
|
/// Starts the thread which will do periodic updates.
|
|
thread = ThreadFromGlobalPool{&PeriodicUpdater::doPeriodicUpdates, this};
|
|
}
|
|
}
|
|
else
|
|
{
|
|
if (thread.joinable())
|
|
{
|
|
/// Wait for the thread to finish.
|
|
auto temp_thread = std::move(thread);
|
|
lock.unlock();
|
|
event.notify_one();
|
|
temp_thread.join();
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
private:
|
|
void doPeriodicUpdates()
|
|
{
|
|
setThreadName("ExterLdrReload");
|
|
|
|
std::unique_lock lock{mutex};
|
|
auto pred = [this] { return !enabled; };
|
|
while (!event.wait_for(lock, std::chrono::seconds(check_period_sec), pred))
|
|
{
|
|
lock.unlock();
|
|
loading_dispatcher.setConfiguration(config_files_reader.read());
|
|
loading_dispatcher.reloadOutdated();
|
|
lock.lock();
|
|
}
|
|
}
|
|
|
|
LoadablesConfigReader & config_files_reader;
|
|
LoadingDispatcher & loading_dispatcher;
|
|
mutable std::mutex mutex;
|
|
bool enabled = false;
|
|
ThreadFromGlobalPool thread;
|
|
std::condition_variable event;
|
|
};
|
|
|
|
|
|
ExternalLoader::ExternalLoader(const String & type_name_, Poco::Logger * log_)
|
|
: config_files_reader(std::make_unique<LoadablesConfigReader>(type_name_, log_))
|
|
, loading_dispatcher(std::make_unique<LoadingDispatcher>(
|
|
[this](auto && a, auto && b, auto && c) { return createObject(a, b, c); },
|
|
type_name_,
|
|
log_))
|
|
, periodic_updater(std::make_unique<PeriodicUpdater>(*config_files_reader, *loading_dispatcher))
|
|
, type_name(type_name_)
|
|
, log(log_)
|
|
{
|
|
}
|
|
|
|
ExternalLoader::~ExternalLoader() = default;
|
|
|
|
scope_guard ExternalLoader::addConfigRepository(std::unique_ptr<IExternalLoaderConfigRepository> repository) const
|
|
{
|
|
auto * ptr = repository.get();
|
|
String name = ptr->getName();
|
|
|
|
config_files_reader->addConfigRepository(std::move(repository));
|
|
reloadConfig(name);
|
|
|
|
return [this, ptr, name]()
|
|
{
|
|
config_files_reader->removeConfigRepository(ptr);
|
|
reloadConfig(name);
|
|
};
|
|
}
|
|
|
|
void ExternalLoader::setConfigSettings(const ExternalLoaderConfigSettings & settings)
|
|
{
|
|
config_files_reader->setConfigSettings(settings);
|
|
}
|
|
|
|
void ExternalLoader::enableAlwaysLoadEverything(bool enable)
|
|
{
|
|
loading_dispatcher->enableAlwaysLoadEverything(enable);
|
|
}
|
|
|
|
void ExternalLoader::enableAsyncLoading(bool enable)
|
|
{
|
|
loading_dispatcher->enableAsyncLoading(enable);
|
|
}
|
|
|
|
void ExternalLoader::enablePeriodicUpdates(bool enable_)
|
|
{
|
|
periodic_updater->enable(enable_);
|
|
}
|
|
|
|
bool ExternalLoader::hasLoadedObjects() const
|
|
{
|
|
return loading_dispatcher->hasLoadedObjects();
|
|
}
|
|
|
|
ExternalLoader::Status ExternalLoader::getCurrentStatus(const String & name) const
|
|
{
|
|
return loading_dispatcher->getCurrentStatus(name);
|
|
}
|
|
|
|
template <typename ReturnType, typename>
|
|
ReturnType ExternalLoader::getLoadResult(const String & name) const
|
|
{
|
|
return loading_dispatcher->getLoadResult<ReturnType>(name);
|
|
}
|
|
|
|
template <typename ReturnType, typename>
|
|
ReturnType ExternalLoader::getLoadResults(const FilterByNameFunction & filter) const
|
|
{
|
|
return loading_dispatcher->getLoadResults<ReturnType>(filter);
|
|
}
|
|
|
|
ExternalLoader::Loadables ExternalLoader::getLoadedObjects() const
|
|
{
|
|
return getLoadResults<Loadables>();
|
|
}
|
|
|
|
ExternalLoader::Loadables ExternalLoader::getLoadedObjects(const FilterByNameFunction & filter) const
|
|
{
|
|
return getLoadResults<Loadables>(filter);
|
|
}
|
|
|
|
size_t ExternalLoader::getNumberOfLoadedObjects() const
|
|
{
|
|
return loading_dispatcher->getNumberOfLoadedObjects();
|
|
}
|
|
|
|
size_t ExternalLoader::getNumberOfObjects() const
|
|
{
|
|
return loading_dispatcher->getNumberOfObjects();
|
|
}
|
|
|
|
template <typename ReturnType, typename>
|
|
ReturnType ExternalLoader::tryLoad(const String & name, Duration timeout) const
|
|
{
|
|
return loading_dispatcher->tryLoad<ReturnType>(name, timeout);
|
|
}
|
|
|
|
template <typename ReturnType, typename>
|
|
ReturnType ExternalLoader::tryLoad(const FilterByNameFunction & filter, Duration timeout) const
|
|
{
|
|
return loading_dispatcher->tryLoad<ReturnType>(filter, timeout);
|
|
}
|
|
|
|
template <typename ReturnType, typename>
|
|
ReturnType ExternalLoader::load(const String & name) const
|
|
{
|
|
auto result = tryLoad<LoadResult>(name);
|
|
checkLoaded(result, false);
|
|
return convertTo<ReturnType>(result);
|
|
}
|
|
|
|
template <typename ReturnType, typename>
|
|
ReturnType ExternalLoader::load(const FilterByNameFunction & filter) const
|
|
{
|
|
auto results = tryLoad<LoadResults>(filter);
|
|
checkLoaded(results, false);
|
|
return convertTo<ReturnType>(results);
|
|
}
|
|
|
|
template <typename ReturnType, typename>
|
|
ReturnType ExternalLoader::loadOrReload(const String & name) const
|
|
{
|
|
loading_dispatcher->setConfiguration(config_files_reader->read());
|
|
auto result = loading_dispatcher->tryLoadOrReload<LoadResult>(name, WAIT);
|
|
checkLoaded(result, true);
|
|
return convertTo<ReturnType>(result);
|
|
}
|
|
|
|
template <typename ReturnType, typename>
|
|
ReturnType ExternalLoader::loadOrReload(const FilterByNameFunction & filter) const
|
|
{
|
|
loading_dispatcher->setConfiguration(config_files_reader->read());
|
|
auto results = loading_dispatcher->tryLoadOrReload<LoadResults>(filter, WAIT);
|
|
checkLoaded(results, true);
|
|
return convertTo<ReturnType>(results);
|
|
}
|
|
|
|
template <typename ReturnType, typename>
|
|
ReturnType ExternalLoader::reloadAllTriedToLoad() const
|
|
{
|
|
std::unordered_set<String> names;
|
|
boost::range::copy(getAllTriedToLoadNames(), std::inserter(names, names.end()));
|
|
return loadOrReload<ReturnType>([&names](const String & name) { return names.count(name); });
|
|
}
|
|
|
|
bool ExternalLoader::has(const String & name) const
|
|
{
|
|
return loading_dispatcher->has(name);
|
|
}
|
|
|
|
Strings ExternalLoader::getAllTriedToLoadNames() const
|
|
{
|
|
return loading_dispatcher->getAllTriedToLoadNames();
|
|
}
|
|
|
|
|
|
void ExternalLoader::checkLoaded(const ExternalLoader::LoadResult & result,
|
|
bool check_no_errors) const
|
|
{
|
|
if (result.object && (!check_no_errors || !result.exception))
|
|
return;
|
|
if (result.status == ExternalLoader::Status::LOADING)
|
|
throw Exception(type_name + " '" + result.name + "' is still loading", ErrorCodes::BAD_ARGUMENTS);
|
|
if (result.exception)
|
|
{
|
|
// Exception is shared for multiple threads.
|
|
// Don't just rethrow it, because sharing the same exception object
|
|
// between multiple threads can lead to weird effects if they decide to
|
|
// modify it, for example, by adding some error context.
|
|
try
|
|
{
|
|
std::rethrow_exception(result.exception);
|
|
}
|
|
catch (const Poco::Exception & e)
|
|
{
|
|
/// This will create a copy for Poco::Exception and DB::Exception
|
|
e.rethrow();
|
|
}
|
|
catch (...)
|
|
{
|
|
throw DB::Exception(ErrorCodes::DICTIONARIES_WAS_NOT_LOADED,
|
|
"Failed to load dictionary '{}': {}",
|
|
result.name,
|
|
getCurrentExceptionMessage(true /*with stack trace*/,
|
|
true /*check embedded stack trace*/));
|
|
}
|
|
}
|
|
if (result.status == ExternalLoader::Status::NOT_EXIST)
|
|
throw Exception(type_name + " '" + result.name + "' not found", ErrorCodes::BAD_ARGUMENTS);
|
|
if (result.status == ExternalLoader::Status::NOT_LOADED)
|
|
throw Exception(type_name + " '" + result.name + "' not tried to load", ErrorCodes::BAD_ARGUMENTS);
|
|
}
|
|
|
|
void ExternalLoader::checkLoaded(const ExternalLoader::LoadResults & results,
|
|
bool check_no_errors) const
|
|
{
|
|
std::exception_ptr exception;
|
|
for (const auto & result : results)
|
|
{
|
|
try
|
|
{
|
|
checkLoaded(result, check_no_errors);
|
|
}
|
|
catch (...)
|
|
{
|
|
if (!exception)
|
|
exception = std::current_exception();
|
|
else
|
|
tryLogCurrentException(log);
|
|
}
|
|
}
|
|
|
|
if (exception)
|
|
std::rethrow_exception(exception);
|
|
}
|
|
|
|
|
|
void ExternalLoader::reloadConfig() const
|
|
{
|
|
loading_dispatcher->setConfiguration(config_files_reader->read());
|
|
}
|
|
|
|
void ExternalLoader::reloadConfig(const String & repository_name) const
|
|
{
|
|
loading_dispatcher->setConfiguration(config_files_reader->read(repository_name));
|
|
}
|
|
|
|
void ExternalLoader::reloadConfig(const String & repository_name, const String & path) const
|
|
{
|
|
loading_dispatcher->setConfiguration(config_files_reader->read(repository_name, path));
|
|
}
|
|
|
|
ExternalLoader::LoadablePtr ExternalLoader::createObject(
|
|
const String & name, const ObjectConfig & config, const LoadablePtr & previous_version) const
|
|
{
|
|
if (previous_version)
|
|
return previous_version->clone();
|
|
|
|
return create(name, *config.config, config.key_in_config, config.repository_name);
|
|
}
|
|
|
|
template ExternalLoader::LoadablePtr ExternalLoader::getLoadResult<ExternalLoader::LoadablePtr>(const String &) const;
|
|
template ExternalLoader::LoadResult ExternalLoader::getLoadResult<ExternalLoader::LoadResult>(const String &) const;
|
|
template ExternalLoader::Loadables ExternalLoader::getLoadResults<ExternalLoader::Loadables>(const FilterByNameFunction &) const;
|
|
template ExternalLoader::LoadResults ExternalLoader::getLoadResults<ExternalLoader::LoadResults>(const FilterByNameFunction &) const;
|
|
|
|
template ExternalLoader::LoadablePtr ExternalLoader::tryLoad<ExternalLoader::LoadablePtr>(const String &, Duration) const;
|
|
template ExternalLoader::LoadResult ExternalLoader::tryLoad<ExternalLoader::LoadResult>(const String &, Duration) const;
|
|
template ExternalLoader::Loadables ExternalLoader::tryLoad<ExternalLoader::Loadables>(const FilterByNameFunction &, Duration) const;
|
|
template ExternalLoader::LoadResults ExternalLoader::tryLoad<ExternalLoader::LoadResults>(const FilterByNameFunction &, Duration) const;
|
|
|
|
template ExternalLoader::LoadablePtr ExternalLoader::load<ExternalLoader::LoadablePtr>(const String &) const;
|
|
template ExternalLoader::LoadResult ExternalLoader::load<ExternalLoader::LoadResult>(const String &) const;
|
|
template ExternalLoader::Loadables ExternalLoader::load<ExternalLoader::Loadables>(const FilterByNameFunction &) const;
|
|
template ExternalLoader::LoadResults ExternalLoader::load<ExternalLoader::LoadResults>(const FilterByNameFunction &) const;
|
|
|
|
template ExternalLoader::LoadablePtr ExternalLoader::loadOrReload<ExternalLoader::LoadablePtr>(const String &) const;
|
|
template ExternalLoader::LoadResult ExternalLoader::loadOrReload<ExternalLoader::LoadResult>(const String &) const;
|
|
template ExternalLoader::Loadables ExternalLoader::loadOrReload<ExternalLoader::Loadables>(const FilterByNameFunction &) const;
|
|
template ExternalLoader::LoadResults ExternalLoader::loadOrReload<ExternalLoader::LoadResults>(const FilterByNameFunction &) const;
|
|
|
|
template ExternalLoader::Loadables ExternalLoader::reloadAllTriedToLoad<ExternalLoader::Loadables>() const;
|
|
template ExternalLoader::LoadResults ExternalLoader::reloadAllTriedToLoad<ExternalLoader::LoadResults>() const;
|
|
}
|