926089746b
The directory utils/ is supposed to contain general-purpose utility
classes and functions, which are either already used across the project,
or are designed to be used across the project.
This patch moves 8 files out of utils/:
utils/advanced_rpc_compressor.hh
utils/advanced_rpc_compressor.cc
utils/advanced_rpc_compressor_protocol.hh
utils/stream_compressor.hh
utils/stream_compressor.cc
utils/dict_trainer.cc
utils/dict_trainer.hh
utils/shared_dict.hh
These 8 files together implement the compression feature of RPC.
None of them are used by any other Scylla component (e.g., sstables have
a different compression), or are ready to be used by another component,
so this patch moves all of them into message/, where RPC is implemented.
Theoretically, we may want in the future to use this cluster of classes
for some other component, but even then, we shouldn't just have these
files individually in utils/ - these are not useful stand-alone
utilities. One cannot use "shared_dict.hh" assuming it is some sort of
general-purpose shared hash table or something - it is completely
specific to compression and zstd, and specifically to its use in those
other classes.
Beyond moving these 8 files, this patch also contains changes to:
1. Fix includes to the 5 moved header files (.hh).
2. Fix configure.py, utils/CMakeLists.txt and message/CMakeLists.txt
for the three moved source files (.cc).
3. In the moved files, change from the "utils::" namespace, to the
"netw::" namespace used by RPC. Also needed to change a bunch
of callers for the new namespace. Also, had to add "utils::"
explicitly in several places which previously assumed the
current namespace is "utils::".
Signed-off-by: Nadav Har'El <nyh@scylladb.com>
Closes scylladb/scylladb#25149
80 lines
2.7 KiB
C++
80 lines
2.7 KiB
C++
/*
|
|
* Copyright (C) 2024-present ScyllaDB
|
|
*/
|
|
|
|
/*
|
|
* SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
|
|
*/
|
|
|
|
#pragma once
|
|
|
|
#include "locator/host_id.hh"
|
|
#include "utils/enum_option.hh"
|
|
#include "utils/updateable_value.hh"
|
|
#include "dict_trainer.hh"
|
|
#include <any>
|
|
|
|
namespace db {
|
|
class system_keyspace;
|
|
} // namespace db
|
|
|
|
namespace utils {
|
|
class alien_worker;
|
|
} // namespace utils
|
|
|
|
namespace service {
|
|
class raft_group0_client;
|
|
class raft_group0;
|
|
} // namespace service
|
|
|
|
namespace gms {
|
|
class feature_service;
|
|
} // namespace gms
|
|
|
|
// A bag of code responsible for starting, stopping, pausing and unpausing RPC compression
|
|
// dictionary training, and for publishing its results to system.dicts (via Raft group 0).
|
|
//
|
|
// It starts the training when the relevant cluster feature is enabled,
|
|
// pauses and unpauses the training appropriately whenever relevant config or leadership status are updated,
|
|
// and publishes new dictionaries whenever the training fiber produces them.
|
|
class dictionary_service {
|
|
db::system_keyspace& _sys_ks;
|
|
locator::host_id _our_host_id;
|
|
utils::updateable_value<enum_option<netw::dict_training_loop::when>> _rpc_dict_training_when;
|
|
service::raft_group0_client& _raft_group0_client;
|
|
abort_source& _as;
|
|
netw::dict_training_loop _training_fiber;
|
|
future<> _training_fiber_future;
|
|
|
|
bool _is_leader = false;
|
|
utils::observer<bool> _leadership_observer;
|
|
utils::observer<enum_option<netw::dict_training_loop::when>> _when_observer;
|
|
std::optional<std::any> _feature_observer;
|
|
|
|
void maybe_toggle_dict_training();
|
|
future<> publish_dict(netw::dict_sampler::dict_type);
|
|
public:
|
|
constexpr static std::string_view rpc_compression_dict_name = "general";
|
|
// This template trick forces the user of `config` to initialize all fields explicitly.
|
|
template <typename Uninitialized = void>
|
|
struct config {
|
|
locator::host_id our_host_id = Uninitialized();
|
|
utils::updateable_value<uint32_t> rpc_dict_training_min_time_seconds = Uninitialized();
|
|
utils::updateable_value<uint64_t> rpc_dict_training_min_bytes = Uninitialized();
|
|
utils::updateable_value<enum_option<netw::dict_training_loop::when>> rpc_dict_training_when = Uninitialized();
|
|
};
|
|
// Note: the training fiber will start as soon as the relevant cluster feature is enabled.
|
|
dictionary_service(
|
|
netw::dict_sampler&,
|
|
db::system_keyspace&,
|
|
utils::alien_worker&,
|
|
service::raft_group0_client&,
|
|
service::raft_group0&,
|
|
abort_source& stop_signal,
|
|
gms::feature_service&,
|
|
config<>
|
|
);
|
|
// For clean shutdown, this must be called and awaited before destruction.
|
|
future<> stop();
|
|
};
|