innov
/
dragonfly
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Switch to using absl flags

This commit is contained in:
Roman Gershman 2022-05-30 06:45:09 +03:00
parent 6dd1552506
commit bb7b205bff
16 changed files with 228 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
helio

@ -1 +1 @@
Subproject commit d74a2550d5d5a587718c8d606dd1797e05db7442
Subproject commit 490814e3f0586e6974a462ddfe42905b71b90c3b

10
src/facade/dragonfly_connection.cc
View File

@ -10,6 +10,7 @@
#include <boost/fiber/operations.hpp>
#include "base/flags.h"
#include "base/logging.h"
#include "facade/conn_context.h"
#include "facade/memcache_parser.h"
@ -19,8 +20,9 @@
#include "util/tls/tls_socket.h"
#include "util/uring/uring_socket.h"
DEFINE_bool(tcp_nodelay, false, "Configures dragonfly connections with socket option TCP_NODELAY");
DEFINE_bool(http_admin_console, true, "If true allows accessing http console on main TCP port");
ABSL_FLAG(bool, tcp_nodelay, false,
"Configures dragonfly connections with socket option TCP_NODELAY");
ABSL_FLAG(bool, http_admin_console, true, "If true allows accessing http console on main TCP port");
using namespace util;
using namespace std;
@ -172,7 +174,7 @@ void Connection::HandleRequests() {
LinuxSocketBase* lsb = static_cast<LinuxSocketBase*>(socket_.get());
if (FLAGS_tcp_nodelay) {
if (absl::GetFlag(FLAGS_tcp_nodelay)) {
int val = 1;
CHECK_EQ(0, setsockopt(lsb->native_handle(), SOL_TCP, TCP_NODELAY, &val, sizeof(val)));
}
@ -194,7 +196,7 @@ void Connection::HandleRequests() {
FiberSocketBase* peer = tls_sock ? (FiberSocketBase*)tls_sock.get() : socket_.get();
io::Result<bool> http_res{false};
if (FLAGS_http_admin_console)
if (absl::GetFlag(FLAGS_http_admin_console))
http_res = CheckForHttpProto(peer);
if (http_res) {

43
src/facade/dragonfly_listener.cc
View File

@ -6,18 +6,21 @@
#include <openssl/ssl.h>
#include "base/flags.h"
#include "base/logging.h"
#include "facade/dragonfly_connection.h"
#include "util/proactor_pool.h"
DEFINE_uint32(conn_threads, 0, "Number of threads used for handing server connections");
DEFINE_bool(tls, false, "");
DEFINE_bool(conn_use_incoming_cpu, false,
"If true uses incoming cpu of a socket in order to distribute"
" incoming connections");
using namespace std;
DEFINE_string(tls_client_cert_file, "", "cert file for tls connections");
DEFINE_string(tls_client_key_file, "", "key file for tls connections");
ABSL_FLAG(uint32_t, conn_threads, 0, "Number of threads used for handing server connections");
ABSL_FLAG(bool, tls, false, "");
ABSL_FLAG(bool, conn_use_incoming_cpu, false,
"If true uses incoming cpu of a socket in order to distribute"
" incoming connections");
ABSL_FLAG(string, tls_client_cert_file, "", "cert file for tls connections");
ABSL_FLAG(string, tls_client_key_file, "", "key file for tls connections");
#if 0
enum TlsClientAuth {
@ -40,14 +43,14 @@ CONFIG_enum(tls_auth_clients, "yes", "", tls_auth_clients_enum, tls_auth_clients
namespace facade {
using namespace util;
using namespace std;
using absl::GetFlag;
namespace {
// To connect: openssl s_client -cipher "ADH:@SECLEVEL=0" -state -crlf -connect 127.0.0.1:6380
static SSL_CTX* CreateSslCntx() {
SSL_CTX* ctx = SSL_CTX_new(TLS_server_method());
if (FLAGS_tls_client_key_file.empty()) {
const auto& tls_client_key_file = GetFlag(FLAGS_tls_client_key_file);
if (tls_client_key_file.empty()) {
// To connect - use openssl s_client -cipher with either:
// "AECDH:@SECLEVEL=0" or "ADH:@SECLEVEL=0" setting.
CHECK_EQ(1, SSL_CTX_set_cipher_list(ctx, "aNULL"));
@ -61,14 +64,14 @@ static SSL_CTX* CreateSslCntx() {
<< "tls-client-key-file not set, no keys are loaded and anonymous ciphers are enabled. "
<< "Do not use in production!";
} else { // tls_client_key_file is set.
CHECK_EQ(1,
SSL_CTX_use_PrivateKey_file(ctx, FLAGS_tls_client_key_file.c_str(), SSL_FILETYPE_PEM));
CHECK_EQ(1, SSL_CTX_use_PrivateKey_file(ctx, tls_client_key_file.c_str(), SSL_FILETYPE_PEM));
const auto& tls_client_cert_file = GetFlag(FLAGS_tls_client_cert_file);
if (!FLAGS_tls_client_cert_file.empty()) {
if (!tls_client_cert_file.empty()) {
// TO connect with redis-cli you need both tls-client-key-file and tls-client-cert-file
// loaded. Use `redis-cli --tls -p 6380 --insecure PING` to test
CHECK_EQ(1, SSL_CTX_use_certificate_chain_file(ctx, FLAGS_tls_client_cert_file.c_str()));
CHECK_EQ(1, SSL_CTX_use_certificate_chain_file(ctx, tls_client_cert_file.c_str()));
}
CHECK_EQ(1, SSL_CTX_set_cipher_list(ctx, "DEFAULT"));
}
@ -99,14 +102,14 @@ bool ConfigureKeepAlive(int fd, unsigned interval_sec) {
return false;
/* Send next probes after the specified interval. Note that we set the
* delay as interval / 3, as we send three probes before detecting
* an error (see the next setsockopt call). */
* delay as interval / 3, as we send three probes before detecting
* an error (see the next setsockopt call). */
val = interval_sec / 3;
if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &val, sizeof(val)) < 0)
return false;
/* Consider the socket in error state after three we send three ACK
* probes without getting a reply. */
* probes without getting a reply. */
val = 3;
if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &val, sizeof(val)) < 0)
return false;
@ -117,7 +120,7 @@ bool ConfigureKeepAlive(int fd, unsigned interval_sec) {
} // namespace
Listener::Listener(Protocol protocol, ServiceInterface* e) : service_(e), protocol_(protocol) {
if (FLAGS_tls) {
if (GetFlag(FLAGS_tls)) {
OPENSSL_init_ssl(OPENSSL_INIT_SSL_DEFAULT, NULL);
ctx_ = CreateSslCntx();
}
@ -157,14 +160,14 @@ void Listener::PostShutdown() {
// We can limit number of threads handling dragonfly connections.
ProactorBase* Listener::PickConnectionProactor(LinuxSocketBase* sock) {
util::ProactorPool* pp = pool();
uint32_t total = FLAGS_conn_threads;
uint32_t total = GetFlag(FLAGS_conn_threads);
uint32_t id = kuint32max;
if (total == 0 || total > pp->size()) {
total = pp->size();
}
if (FLAGS_conn_use_incoming_cpu) {
if (GetFlag(FLAGS_conn_use_incoming_cpu)) {
int fd = sock->native_handle();
int cpu, napi_id;

7
src/facade/ok_main.cc
View File

@ -9,10 +9,11 @@
#include "util/accept_server.h"
#include "util/uring/uring_pool.h"
DEFINE_uint32(port, 6379, "server port");
ABSL_FLAG(uint32_t, port, 6379, "server port");
using namespace util;
using namespace std;
using absl::GetFlag;
namespace facade {
@ -43,7 +44,7 @@ class OkService : public ServiceInterface {
void RunEngine(ProactorPool* pool, AcceptServer* acceptor) {
OkService service;
acceptor->AddListener(FLAGS_port, new Listener{Protocol::REDIS, &service});
acceptor->AddListener(GetFlag(FLAGS_port), new Listener{Protocol::REDIS, &service});
acceptor->Run();
acceptor->Wait();
@ -56,7 +57,7 @@ void RunEngine(ProactorPool* pool, AcceptServer* acceptor) {
int main(int argc, char* argv[]) {
MainInitGuard guard(&argc, &argv);
CHECK_GT(FLAGS_port, 0u);
CHECK_GT(GetFlag(FLAGS_port), 0u);
uring::UringPool pp{1024};
pp.Run();

13
src/server/debugcmd.cc
View File

@ -9,6 +9,7 @@
#include <boost/fiber/operations.hpp>
#include <filesystem>
#include "base/flags.h"
#include "base/logging.h"
#include "server/blocking_controller.h"
#include "server/engine_shard_set.h"
@ -20,12 +21,13 @@
#include "server/transaction.h"
#include "util/uring/uring_fiber_algo.h"
DECLARE_string(dir);
DECLARE_string(dbfilename);
using namespace std;
ABSL_DECLARE_FLAG(string, dir);
ABSL_DECLARE_FLAG(string, dbfilename);
namespace dfly {
using namespace std;
using namespace util;
namespace this_fiber = ::boost::this_fiber;
using boost::intrusive_ptr;
@ -33,6 +35,7 @@ using boost::fibers::fiber;
using namespace facade;
namespace fs = std::filesystem;
using absl::StrAppend;
using absl::GetFlag;
struct PopulateBatch {
DbIndex dbid;
@ -181,8 +184,8 @@ void DebugCmd::Load(std::string_view filename) {
fs::path path(filename);
if (filename.empty()) {
fs::path dir_path(FLAGS_dir);
string filename = FLAGS_dbfilename;
fs::path dir_path(GetFlag(FLAGS_dir));
string filename = GetFlag(FLAGS_dbfilename);
dir_path.append(filename);
path = dir_path;
}

117
src/server/dfly_main.cc
View File

@ -6,6 +6,11 @@
#include <mimalloc-new-delete.h>
#endif
#include <absl/flags/usage.h>
#include <absl/flags/usage_config.h>
#include <absl/strings/match.h>
#include <absl/strings/str_cat.h>
#include <absl/strings/strip.h>
#include <mimalloc.h>
#include "base/init.h"
@ -18,26 +23,80 @@
#include "util/uring/uring_pool.h"
#include "util/varz.h"
using namespace std;
DECLARE_uint32(port);
DECLARE_uint32(dbnum);
DECLARE_uint32(memcache_port);
DECLARE_uint64(maxmemory);
ABSL_DECLARE_FLAG(uint32_t, port);
ABSL_DECLARE_FLAG(uint32_t, dbnum);
ABSL_DECLARE_FLAG(uint32_t, memcache_port);
ABSL_DECLARE_FLAG(uint64_t, maxmemory);
DEFINE_bool(use_large_pages, false, "If true - uses large memory pages for allocations");
DEFINE_string(bind, "",
"Bind address. If empty - binds on all interfaces. "
"It's not advised due to security implications.");
ABSL_FLAG(bool, use_large_pages, false, "If true - uses large memory pages for allocations");
ABSL_FLAG(string, bind, "",
"Bind address. If empty - binds on all interfaces. "
"It's not advised due to security implications.");
using namespace util;
using namespace std;
using namespace facade;
using absl::GetFlag;
using absl::StrCat;
using strings::HumanReadableNumBytes;
namespace dfly {
namespace {
enum class TermColor { kDefault, kRed, kGreen, kYellow };
// Returns the ANSI color code for the given color. TermColor::kDefault is
// an invalid input.
static const char* GetAnsiColorCode(TermColor color) {
switch (color) {
case TermColor::kRed:
return "1";
case TermColor::kGreen:
return "2";
case TermColor::kYellow:
return "3";
default:
return nullptr;
}
}
string ColorStart(TermColor color) {
return StrCat("\033[0;3", GetAnsiColorCode(color), "m");
}
// Resets the terminal to default.
const char kColorEnd[] = "\033[m";
string ColoredStr(TermColor color, string_view str) {
return StrCat(ColorStart(color), str, kColorEnd);
}
bool HelpshortFlags(std::string_view f) {
return absl::StartsWith(f, "\033[0;32");
}
bool HelpFlags(std::string_view f) {
return absl::StartsWith(f, "\033[0;3");
}
string NormalizePaths(std::string_view path) {
if (absl::ConsumePrefix(&path, "../src/"))
return ColoredStr(TermColor::kGreen, path);
if (absl::ConsumePrefix(&path, "../"))
return ColoredStr(TermColor::kYellow, path);
if (absl::ConsumePrefix(&path, "_deps/"))
return string(path);
return string(path);
}
bool RunEngine(ProactorPool* pool, AcceptServer* acceptor) {
if (FLAGS_maxmemory > 0 && FLAGS_maxmemory < pool->size() * 256_MB) {
auto maxmemory = GetFlag(FLAGS_maxmemory);
if (maxmemory > 0 && maxmemory < pool->size() * 256_MB) {
LOG(ERROR) << "Max memory is less than 256MB per thread. Exiting...";
return false;
}
@ -49,15 +108,17 @@ bool RunEngine(ProactorPool* pool, AcceptServer* acceptor) {
Service::InitOpts opts;
opts.disable_time_update = false;
service.Init(acceptor, main_listener, opts);
const auto& bind = GetFlag(FLAGS_bind);
const char* bind_addr = bind.empty() ? nullptr : bind.c_str();
auto port = GetFlag(FLAGS_port);
auto mc_port = GetFlag(FLAGS_memcache_port);
const char* bind_addr = FLAGS_bind.empty() ? nullptr : FLAGS_bind.c_str();
error_code ec = acceptor->AddListener(bind_addr, port, main_listener);
error_code ec = acceptor->AddListener(bind_addr, FLAGS_port, main_listener);
LOG_IF(FATAL, ec) << "Cound not open port " << port << ", error: " << ec.message();
LOG_IF(FATAL, ec) << "Cound not open port " << FLAGS_port << ", error: " << ec.message();
if (FLAGS_memcache_port > 0) {
acceptor->AddListener(FLAGS_memcache_port, new Listener{Protocol::MEMCACHE, &service});
if (mc_port > 0) {
acceptor->AddListener(mc_port, new Listener{Protocol::MEMCACHE, &service});
}
acceptor->Run();
@ -67,17 +128,23 @@ bool RunEngine(ProactorPool* pool, AcceptServer* acceptor) {
return true;
}
} // namespace
} // namespace dfly
extern "C" void _mi_options_init();
int main(int argc, char* argv[]) {
gflags::SetUsageMessage("dragonfly [FLAGS]");
gflags::SetVersionString("v0.2");
absl::SetProgramUsageMessage("dragonfly [FLAGS]");
absl::FlagsUsageConfig config;
config.contains_help_flags = dfly::HelpFlags;
config.contains_helpshort_flags = dfly::HelpshortFlags;
config.normalize_filename = dfly::NormalizePaths;
absl::SetFlagsUsageConfig(config);
MainInitGuard guard(&argc, &argv);
CHECK_GT(FLAGS_port, 0u);
CHECK_GT(GetFlag(FLAGS_port), 0u);
mi_stats_reset();
base::sys::KernelVersion kver;
@ -88,7 +155,7 @@ int main(int argc, char* argv[]) {
return 1;
}
if (FLAGS_dbnum > dfly::kMaxDbId) {
if (GetFlag(FLAGS_dbnum) > dfly::kMaxDbId) {
LOG(ERROR) << "dbnum is too big. Exiting...";
return 1;
}
@ -96,7 +163,7 @@ int main(int argc, char* argv[]) {
CHECK_LT(kver.major, 99u);
dfly::kernel_version = kver.kernel * 100 + kver.major;
if (FLAGS_maxmemory == 0) {
if (GetFlag(FLAGS_maxmemory) == 0) {
LOG(INFO) << "maxmemory has not been specified. Deciding myself....";
io::Result<io::MemInfoData> res = io::ReadMemInfo();
@ -104,14 +171,14 @@ int main(int argc, char* argv[]) {
size_t maxmemory = size_t(0.8 * available);
LOG(INFO) << "Found " << HumanReadableNumBytes(available)
<< " available memory. Setting maxmemory to " << HumanReadableNumBytes(maxmemory);
FLAGS_maxmemory = maxmemory;
absl::SetFlag(&FLAGS_maxmemory, maxmemory);
} else {
LOG(INFO) << "Max memory limit is: " << HumanReadableNumBytes(FLAGS_maxmemory);
LOG(INFO) << "Max memory limit is: " << HumanReadableNumBytes(GetFlag(FLAGS_maxmemory));
}
dfly::max_memory_limit = FLAGS_maxmemory;
dfly::max_memory_limit = GetFlag(FLAGS_maxmemory);
if (FLAGS_use_large_pages) {
if (GetFlag(FLAGS_use_large_pages)) {
mi_option_enable(mi_option_large_os_pages);
}
mi_option_enable(mi_option_show_errors);

15
src/server/engine_shard_set.cc
View File

@ -9,6 +9,7 @@ extern "C" {
#include "redis/zmalloc.h"
}
#include "base/flags.h"
#include "base/logging.h"
#include "server/blocking_controller.h"
#include "server/server_state.h"
@ -17,12 +18,13 @@ extern "C" {
#include "util/fiber_sched_algo.h"
#include "util/varz.h"
DEFINE_string(backing_prefix, "", "");
DECLARE_bool(cache_mode);
using namespace std;
ABSL_FLAG(string, backing_prefix, "", "");
ABSL_DECLARE_FLAG(bool, cache_mode);
namespace dfly {
using namespace std;
using namespace util;
namespace this_fiber = ::boost::this_fiber;
namespace fibers = ::boost::fibers;
@ -46,7 +48,7 @@ EngineShard::Stats& EngineShard::Stats::operator+=(const EngineShard::Stats& o)
EngineShard::EngineShard(util::ProactorBase* pb, bool update_db_time, mi_heap_t* heap)
: queue_(kQueueLen), txq_([](const Transaction* t) { return t->txid(); }), mi_resource_(heap),
db_slice_(pb->GetIndex(), FLAGS_cache_mode, this) {
db_slice_(pb->GetIndex(), absl::GetFlag(FLAGS_cache_mode), this) {
fiber_q_ = fibers::fiber([this, index = pb->GetIndex()] {
this_fiber::properties<FiberProps>().set_name(absl::StrCat("shard_queue", index));
queue_.Run();
@ -90,9 +92,10 @@ void EngineShard::InitThreadLocal(ProactorBase* pb, bool update_db_time) {
CompactObj::InitThreadLocal(shard_->memory_resource());
SmallString::InitThreadLocal(data_heap);
if (!FLAGS_backing_prefix.empty()) {
string backing_prefix = absl::GetFlag(FLAGS_backing_prefix);
if (!backing_prefix.empty()) {
string fn =
absl::StrCat(FLAGS_backing_prefix, "-", absl::Dec(pb->GetIndex(), absl::kZeroPad4), ".ssd");
absl::StrCat(backing_prefix, "-", absl::Dec(pb->GetIndex(), absl::kZeroPad4), ".ssd");
shard_->tiered_storage_.reset(new TieredStorage(&shard_->db_slice_));
error_code ec = shard_->tiered_storage_->Open(fn);

11
src/server/generic_family.cc
View File

@ -9,6 +9,7 @@ extern "C" {
#include "redis/util.h"
}
#include "base/flags.h"
#include "base/logging.h"
#include "server/blocking_controller.h"
#include "server/command_registry.h"
@ -18,8 +19,8 @@ extern "C" {
#include "server/transaction.h"
#include "util/varz.h"
DEFINE_uint32(dbnum, 16, "Number of databases");
DEFINE_uint32(keys_output_limit, 8192, "Maximum number of keys output by keys command");
ABSL_FLAG(uint32_t, dbnum, 16, "Number of databases");
ABSL_FLAG(uint32_t, keys_output_limit, 8192, "Maximum number of keys output by keys command");
namespace dfly {
using namespace std;
@ -400,9 +401,11 @@ void GenericFamily::Keys(CmdArgList args, ConnectionContext* cntx) {
ScanOpts scan_opts;
scan_opts.pattern = pattern;
scan_opts.limit = 512;
auto output_limit = absl::GetFlag(FLAGS_keys_output_limit);
do {
cursor = ScanGeneric(cursor, scan_opts, &keys, cntx);
} while (cursor != 0 && keys.size() < FLAGS_keys_output_limit);
} while (cursor != 0 && keys.size() < output_limit);
(*cntx)->StartArray(keys.size());
for (const auto& k : keys) {
@ -481,7 +484,7 @@ void GenericFamily::Select(CmdArgList args, ConnectionContext* cntx) {
if (!absl::SimpleAtoi(key, &index)) {
return (*cntx)->SendError(kInvalidDbIndErr);
}
if (index < 0 || index >= FLAGS_dbnum) {
if (index < 0 || index >= absl::GetFlag(FLAGS_dbnum)) {
return (*cntx)->SendError(kDbIndOutOfRangeErr);
}
cntx->conn_state.db_index = index;

7
src/server/io_mgr.cc
View File

@ -7,11 +7,12 @@
#include <fcntl.h>
#include <mimalloc.h>
#include "base/flags.h"
#include "base/logging.h"
#include "facade/facade_types.h"
#include "util/uring/proactor.h"
DEFINE_bool(backing_file_direct, false, "If true uses O_DIRECT to open backing files");
ABSL_FLAG(bool, backing_file_direct, false, "If true uses O_DIRECT to open backing files");
namespace dfly {
@ -41,7 +42,7 @@ error_code IoMgr::Open(const string& path) {
CHECK(!backing_file_);
int kFlags = O_CREAT | O_RDWR | O_TRUNC | O_CLOEXEC;
if (FLAGS_backing_file_direct) {
if (absl::GetFlag(FLAGS_backing_file_direct)) {
kFlags |= O_DIRECT;
}
auto res = uring::OpenLinux(path, kFlags, 0666);
@ -111,7 +112,7 @@ error_code IoMgr::WriteAsync(size_t offset, string_view blob, WriteCb cb) {
error_code IoMgr::Read(size_t offset, io::MutableBytes dest) {
DCHECK(!dest.empty());
if (FLAGS_backing_file_direct) {
if (absl::GetFlag(FLAGS_backing_file_direct)) {
size_t read_offs = offset & ~4095ULL;
size_t end_range = alignup(offset + dest.size(), 4096);
size_t space_needed = end_range - read_offs;

12
src/server/list_family.cc
View File

@ -10,6 +10,7 @@ extern "C" {
#include <absl/strings/numbers.h>
#include "base/flags.h"
#include "base/logging.h"
#include "server/blocking_controller.h"
#include "server/command_registry.h"
@ -34,7 +35,7 @@ extern "C" {
* but if your use case is unique, adjust the settings as necessary.
*
*/
DEFINE_int32(list_max_listpack_size, -2, "Maximum listpack size, default is 8kb");
ABSL_FLAG(int32_t, list_max_listpack_size, -2, "Maximum listpack size, default is 8kb");
/**
* Lists may also be compressed.
@ -54,12 +55,13 @@ DEFINE_int32(list_max_listpack_size, -2, "Maximum listpack size, default is 8kb"
*
*/
DEFINE_int32(list_compress_depth, 0, "Compress depth of the list. Default is no compression");
ABSL_FLAG(int32_t, list_compress_depth, 0, "Compress depth of the list. Default is no compression");
namespace dfly {
using namespace std;
using namespace facade;
using absl::GetFlag;
namespace {
@ -316,7 +318,8 @@ OpResult<string> OpRPopLPushSingleShard(const OpArgs& op_args, string_view src,
if (res.second) {
robj* obj = createQuicklistObject();
dest_ql = (quicklist*)obj->ptr;
quicklistSetOptions(dest_ql, FLAGS_list_max_listpack_size, FLAGS_list_compress_depth);
quicklistSetOptions(dest_ql, GetFlag(FLAGS_list_max_listpack_size),
GetFlag(FLAGS_list_compress_depth));
dest_it->second.ImportRObj(obj);
// Insertion of dest could invalidate src_it. Find it again.
@ -391,7 +394,8 @@ OpResult<uint32_t> OpPush(const OpArgs& op_args, std::string_view key, ListDir d
if (new_key) {
robj* o = createQuicklistObject();
ql = (quicklist*)o->ptr;
quicklistSetOptions(ql, FLAGS_list_max_listpack_size, FLAGS_list_compress_depth);
quicklistSetOptions(ql, GetFlag(FLAGS_list_max_listpack_size),
GetFlag(FLAGS_list_compress_depth));
it->second.ImportRObj(o);
} else {
if (it->second.ObjType() != OBJ_LIST)

20
src/server/main_service.cc
View File

@ -16,6 +16,7 @@ extern "C" {
#include <boost/fiber/operations.hpp>
#include <filesystem>
#include "base/flags.h"
#include "base/logging.h"
#include "facade/dragonfly_connection.h"
#include "facade/error.h"
@ -34,22 +35,24 @@ extern "C" {
#include "util/uring/uring_fiber_algo.h"
#include "util/varz.h"
using namespace std;
// TODO: to move to absl flags and keep legacy flags only for glog library.
// absl flags allow parsing of custom types and allow specifying which flags appear
// for helpshort.
DEFINE_uint32(port, 6379, "Redis port");
DEFINE_uint32(memcache_port, 0, "Memcached port");
DECLARE_string(requirepass);
DEFINE_uint64(maxmemory, 0,
ABSL_FLAG(uint32_t, port, 6379, "Redis port");
ABSL_FLAG(uint32_t, memcache_port, 0, "Memcached port");
ABSL_FLAG(uint64_t, maxmemory, 0,
"Limit on maximum-memory that is used by the database."
"0 - means the program will automatically determine its maximum memory usage");
DEFINE_bool(cache_mode, false,
ABSL_FLAG(bool, cache_mode, false,
"If true, the backend behaves like a cache, "
"by evicting entries when getting close to maxmemory limit");
ABSL_DECLARE_FLAG(string, requirepass);
namespace dfly {
using namespace std;
using namespace util;
using base::VarzValue;
using ::boost::intrusive_ptr;
@ -57,6 +60,7 @@ namespace fibers = ::boost::fibers;
namespace this_fiber = ::boost::this_fiber;
using facade::MCReplyBuilder;
using facade::RedisReplyBuilder;
using absl::GetFlag;
namespace {
@ -564,7 +568,7 @@ void Service::DispatchMC(const MemcacheParser::Command& cmd, std::string_view va
server_family_.StatsMC(cmd.key, cntx);
return;
case MemcacheParser::VERSION:
mc_builder->SendSimpleString(absl::StrCat("VERSION ", gflags::VersionString()));
mc_builder->SendSimpleString(absl::StrCat("VERSION ", "TBD"));
return;
default:
mc_builder->SendClientError("bad command line format");
@ -654,7 +658,7 @@ bool Service::IsShardSetLocked() const {
}
bool Service::IsPassProtected() const {
return !FLAGS_requirepass.empty();
return !GetFlag(FLAGS_requirepass).empty();
}
absl::flat_hash_map<std::string, unsigned> Service::UknownCmdMap() const {

16
src/server/rdb_load.cc
View File

@ -20,6 +20,7 @@ extern "C" {
#include <absl/strings/str_cat.h>
#include "base/endian.h"
#include "base/flags.h"
#include "base/logging.h"
#include "server/engine_shard_set.h"
#include "server/error.h"
@ -29,9 +30,9 @@ extern "C" {
#include "server/set_family.h"
#include "strings/human_readable.h"
DECLARE_int32(list_max_listpack_size);
DECLARE_int32(list_compress_depth);
DECLARE_uint32(dbnum);
ABSL_DECLARE_FLAG(int32_t, list_max_listpack_size);
ABSL_DECLARE_FLAG(int32_t, list_compress_depth);
ABSL_DECLARE_FLAG(uint32_t, dbnum);
namespace dfly {
@ -39,6 +40,7 @@ using namespace std;
using base::IoBuf;
using nonstd::make_unexpected;
using namespace util;
using absl::GetFlag;
using rdb::errc;
namespace {
@ -190,8 +192,7 @@ struct RdbLoader::ObjSettings {
ObjSettings() = default;
};
RdbLoader::RdbLoader(ScriptMgr* script_mgr)
: script_mgr_(script_mgr), mem_buf_{16_KB} {
RdbLoader::RdbLoader(ScriptMgr* script_mgr) : script_mgr_(script_mgr), mem_buf_{16_KB} {
shard_buf_.reset(new ItemsBuf[shard_set->size()]);
}
@ -302,7 +303,7 @@ error_code RdbLoader::Load(io::Source* src) {
/* SELECTDB: Select the specified database. */
SET_OR_RETURN(LoadLen(nullptr), dbid);
if (dbid > FLAGS_dbnum) {
if (dbid > GetFlag(FLAGS_dbnum)) {
LOG(WARNING) << "database id " << dbid << " exceeds dbnum limit. Try increasing the flag.";
return RdbError(errc::bad_db_index);
@ -1193,7 +1194,8 @@ io::Result<robj*> RdbLoader::ReadListQuicklist(int rdbtype) {
if (len == 0)
return Unexpected(errc::empty_key);
quicklist* ql = quicklistNew(FLAGS_list_max_listpack_size, FLAGS_list_compress_depth);
quicklist* ql =
quicklistNew(GetFlag(FLAGS_list_max_listpack_size), GetFlag(FLAGS_list_compress_depth));
uint64_t container = QUICKLIST_NODE_CONTAINER_PACKED;
auto cleanup = absl::Cleanup([&] { quicklistRelease(ql); });

15
src/server/rdb_test.cc
View File

@ -8,8 +8,11 @@ extern "C" {
#include "redis/zmalloc.h"
}
#include <absl/flags/reflection.h>
#include <mimalloc.h>
#include "base/flags.h"
#include "base/gtest.h"
#include "base/logging.h"
#include "facade/facade_test.h" // needed to find operator== for RespExpr.
@ -24,9 +27,10 @@ using namespace std;
using namespace util;
using namespace facade;
using absl::StrCat;
using absl::SetFlag;
DECLARE_int32(list_compress_depth);
DECLARE_int32(list_max_listpack_size);
ABSL_DECLARE_FLAG(int32, list_compress_depth);
ABSL_DECLARE_FLAG(int32, list_max_listpack_size);
namespace dfly {
@ -109,9 +113,10 @@ TEST_F(RdbTest, LoadSmall6) {
}
TEST_F(RdbTest, Reload) {
gflags::FlagSaver fs;
FLAGS_list_compress_depth = 1;
FLAGS_list_max_listpack_size = 1; // limit listpack to a single element.
absl::FlagSaver fs;
SetFlag(&FLAGS_list_compress_depth, 1);
SetFlag(&FLAGS_list_max_listpack_size, 1); // limit listpack to a single element.
Run({"set", "string_key", "val"});
Run({"set", "large_key", string(511, 'L')});

37
src/server/server_family.cc
View File

@ -17,6 +17,7 @@ extern "C" {
#include "redis/redis_aux.h"
}
#include "base/flags.h"
#include "base/logging.h"
#include "facade/dragonfly_connection.h"
#include "io/file_util.h"
@ -38,24 +39,26 @@ extern "C" {
#include "util/accept_server.h"
#include "util/uring/uring_file.h"
DEFINE_string(dir, "", "working directory");
DEFINE_string(dbfilename, "dump", "the filename to save/load the DB");
DEFINE_string(requirepass, "", "password for AUTH authentication");
using namespace std;
DECLARE_uint32(port);
DECLARE_bool(cache_mode);
ABSL_FLAG(string, dir, "", "working directory");
ABSL_FLAG(string, dbfilename, "dump", "the filename to save/load the DB");
ABSL_FLAG(string, requirepass, "", "password for AUTH authentication");
ABSL_DECLARE_FLAG(uint32_t, port);
ABSL_DECLARE_FLAG(bool, cache_mode);
extern "C" mi_stats_t _mi_stats_main;
namespace dfly {
using namespace std;
using namespace util;
namespace fibers = ::boost::fibers;
namespace fs = std::filesystem;
using absl::StrCat;
using facade::MCReplyBuilder;
using strings::HumanReadableNumBytes;
using absl::GetFlag;
namespace {
@ -85,10 +88,12 @@ string UnknownSubCmd(string_view subcmd, string cmd) {
}
string InferLoadFile(fs::path data_dir) {
if (FLAGS_dbfilename.empty())
const auto& dbname = GetFlag(FLAGS_dbfilename);
if (dbname.empty())
return string{};
fs::path fl_path = data_dir.append(FLAGS_dbfilename);
fs::path fl_path = data_dir.append(dbname);
if (fs::exists(fl_path))
return fl_path.generic_string();
@ -167,9 +172,10 @@ void ServerFamily::Init(util::AcceptServer* acceptor, util::ListenerInterface* m
task_10ms_ = pb_task_->AwaitBrief([&] { return pb_task_->AddPeriodic(10, cache_cb); });
fs::path data_folder = fs::current_path();
const auto& dir = GetFlag(FLAGS_dir);
if (!FLAGS_dir.empty()) {
data_folder = FLAGS_dir;
if (!dir.empty()) {
data_folder = dir;
error_code ec;
@ -306,7 +312,7 @@ void ServerFamily::StatsMC(std::string_view section, facade::ConnectionContext*
}
error_code ServerFamily::DoSave(Transaction* trans, string* err_details) {
fs::path dir_path(FLAGS_dir);
fs::path dir_path(GetFlag(FLAGS_dir));
error_code ec;
if (!dir_path.empty()) {
@ -317,7 +323,8 @@ error_code ServerFamily::DoSave(Transaction* trans, string* err_details) {
}
}
fs::path filename = FLAGS_dbfilename.empty() ? "dump" : FLAGS_dbfilename;
const auto& dbfilename = GetFlag(FLAGS_dbfilename);
fs::path filename = dbfilename.empty() ? "dump" : dbfilename;
fs::path path = dir_path;
if (!filename.has_extension()) {
@ -471,7 +478,7 @@ void ServerFamily::Auth(CmdArgList args, ConnectionContext* cntx) {
}
string_view pass = ArgS(args, 1);
if (pass == FLAGS_requirepass) {
if (pass == GetFlag(FLAGS_requirepass)) {
cntx->authenticated = true;
(*cntx)->SendOk();
} else {
@ -644,7 +651,7 @@ void ServerFamily::Info(CmdArgList args, ConnectionContext* cntx) {
append("redis_mode", "standalone");
append("arch_bits", 64);
append("multiplexing_api", "iouring");
append("tcp_port", FLAGS_port);
append("tcp_port", GetFlag(FLAGS_port));
size_t uptime = time(NULL) - start_time_;
append("uptime_in_seconds", uptime);
@ -697,7 +704,7 @@ void ServerFamily::Info(CmdArgList args, ConnectionContext* cntx) {
append("small_string_bytes", m.small_string_bytes);
append("maxmemory", max_memory_limit);
append("maxmemory_human", HumanReadableNumBytes(max_memory_limit));
append("cache_mode", FLAGS_cache_mode ? "cache" : "store");
append("cache_mode", GetFlag(FLAGS_cache_mode) ? "cache" : "store");
}
if (should_enter("STATS")) {

8
src/server/test_utils.cc
View File

@ -12,16 +12,18 @@ extern "C" {
#include <absl/strings/str_split.h>
#include <mimalloc.h>
#include "base/flags.h"
#include "base/logging.h"
#include "base/stl_util.h"
#include "facade/dragonfly_connection.h"
#include "util/uring/uring_pool.h"
DECLARE_string(dbfilename);
using namespace std;
ABSL_DECLARE_FLAG(string, dbfilename);
namespace dfly {
using MP = MemcacheParser;
using namespace std;
using namespace util;
using namespace testing;
@ -52,7 +54,7 @@ BaseFamilyTest::~BaseFamilyTest() {
}
void BaseFamilyTest::SetUpTestSuite() {
FLAGS_dbfilename = "";
absl::SetFlag(&FLAGS_dbfilename, "");
init_zmalloc_threadlocal(mi_heap_get_backing());
}

13
src/server/tiered_storage.cc
View File

@ -10,14 +10,17 @@ extern "C" {
#include <mimalloc.h>
#include "base/flags.h"
#include "base/logging.h"
#include "server/db_slice.h"
#include "util/proactor_base.h"
DEFINE_uint32(tiered_storage_max_pending_writes, 32, "Maximal number of pending writes per thread");
ABSL_FLAG(uint32_t, tiered_storage_max_pending_writes, 32,
"Maximal number of pending writes per thread");
namespace dfly {
using namespace std;
using absl::GetFlag;
struct IndexKey {
DbIndex db_indx;
@ -181,7 +184,7 @@ void TieredStorage::SendIoRequest(ActiveIoRequest* req) {
string_view sv{req->block_ptr, kBatchSize};
active_req_sem_.await(
[this] { return num_active_requests_ <= FLAGS_tiered_storage_max_pending_writes; });
[this] { return num_active_requests_ <= GetFlag(FLAGS_tiered_storage_max_pending_writes); });
auto cb = [this, req](int res) { FinishIoRequest(res, req); };
++num_active_requests_;
@ -230,7 +233,7 @@ void TieredStorage::FinishIoRequest(int io_res, ActiveIoRequest* req) {
delete req;
--num_active_requests_;
if (num_active_requests_ == FLAGS_tiered_storage_max_pending_writes) {
if (num_active_requests_ == GetFlag(FLAGS_tiered_storage_max_pending_writes)) {
active_req_sem_.notifyAll();
}
@ -253,7 +256,7 @@ void TieredStorage::SetExternal(DbIndex db_index, size_t item_offset, PrimeValue
}
bool TieredStorage::ShouldFlush() {
if (num_active_requests_ >= FLAGS_tiered_storage_max_pending_writes)
if (num_active_requests_ >= GetFlag(FLAGS_tiered_storage_max_pending_writes))
return false;
return pending_req_.size() > pending_req_.capacity() / 2;
@ -264,7 +267,7 @@ error_code TieredStorage::UnloadItem(DbIndex db_index, PrimeIterator it) {
size_t blob_len = it->second.Size();
if (blob_len >= kBatchSize / 2 &&
num_active_requests_ < FLAGS_tiered_storage_max_pending_writes) {
num_active_requests_ < GetFlag(FLAGS_tiered_storage_max_pending_writes)) {
LOG(FATAL) << "TBD";
}
error_code ec;