use super::args::{self, Args, is};
use super::table::{self, Spec};
use super::{DATABASES, Flow, Server, Session, cpu};
use crate::proto::Proto;
use crate::reply::Out;
use core::fmt::Write;
use std::time::{SystemTime, UNIX_EPOCH};
use yo_common::num::parse_i64;
use yo_common::{Code, Error, Result, glob};
use yo_kv::Keyspace;
const REPORTED_SERVER: &str = "redis";
const REPORTED_VERSION: &str = "8.8.0";
const SETTINGS: &[(&str, &str)] = &[
("appendonly", "no"),
("appendfsync", "everysec"),
("databases", "16"),
("io-threads", "1"),
("maxmemory", "0"),
("maxmemory-policy", "noeviction"),
("proto-max-bulk-len", "536870912"),
("save", ""),
("timeout", "0"),
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Knob {
SetIntsetEntries,
SetListpackEntries,
SetListpackValue,
HashListpackEntries,
HashListpackValue,
}
const LADDER: &[(&str, Knob)] = &[
("hash-max-listpack-entries", Knob::HashListpackEntries),
("hash-max-listpack-value", Knob::HashListpackValue),
("hash-max-ziplist-entries", Knob::HashListpackEntries),
("hash-max-ziplist-value", Knob::HashListpackValue),
("set-max-intset-entries", Knob::SetIntsetEntries),
("set-max-listpack-entries", Knob::SetListpackEntries),
("set-max-listpack-value", Knob::SetListpackValue),
];
pub(super) fn execute(
server: &mut Server,
session: &mut Session,
spec: &Spec,
args: Args<'_>,
out: &mut Out,
) -> Result<Flow> {
match spec.name {
"ping" => {
if args.len() > 2 {
return Err(args::wrong_arity("ping"));
}
if args.len() == 2 {
out.bulk(args.get(1));
} else {
out.simple(b"PONG");
}
}
"echo" => out.bulk(args.get(1)),
"hello" => hello(session, args, out)?,
"select" => {
let n = args.int(1)?;
let ok = usize::try_from(n).is_ok_and(|n| n < DATABASES);
if !ok {
return Err(Error::new(Code::Invalid, "DB index is out of range"));
}
session.db = n as usize;
out.ok();
}
"reset" => {
session.reset();
out.set_proto(Proto::Resp2);
out.simple(b"RESET");
}
"quit" => {
out.ok();
return Ok(Flow::Close);
}
"command" => command(args, out)?,
"config" => config(server, args, out)?,
"info" => info(server, args, out),
"dbsize" => out.int(server.dbs[session.db].len() as i64),
"flushall" => {
flush_mode(args)?;
for db in &mut server.dbs {
db.clear();
}
out.ok();
}
"flushdb" => {
flush_mode(args)?;
server.dbs[session.db].clear();
out.ok();
}
"time" => time(out),
_ => return Err(args::unknown_command(args)),
}
Ok(Flow::Continue)
}
fn time(out: &mut Out) {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default();
out.array(2);
out.bulk(now.as_secs().to_string().as_bytes());
out.bulk(now.subsec_micros().to_string().as_bytes());
}
fn flush_mode(args: Args<'_>) -> Result<()> {
if args.len() == 1 {
return Ok(());
}
if args.len() > 2 || !(is(args.get(1), b"async") || is(args.get(1), b"sync")) {
return Err(args::syntax());
}
Ok(())
}
fn hello(session: &mut Session, args: Args<'_>, out: &mut Out) -> Result<()> {
if args.len() > 1 {
let v = parse_i64(args.get(1)).ok_or_else(|| {
Error::new(
Code::Invalid,
"Protocol version is not an integer or out of range",
)
})?;
let Some(proto) = Proto::from_version(v) else {
out.error(b"NOPROTO unsupported protocol version");
return Ok(());
};
let mut i = 2;
while i < args.len() {
let o = args.get(i);
if is(o, b"AUTH") && i + 2 < args.len() {
if !is(args.get(i + 1), b"default") {
out.error(b"WRONGPASS invalid username-password pair or user is disabled.");
return Ok(());
}
i += 3;
} else if is(o, b"SETNAME") && i + 1 < args.len() {
session.set_name(args.get(i + 1));
i += 2;
} else {
return Err(yo_alloc::allow(|| {
Error::fmt(
Code::Invalid,
format_args!(
"Syntax error in HELLO option '{}'",
String::from_utf8_lossy(o)
),
)
}));
}
}
out.set_proto(proto);
}
let proto = out.proto().version();
out.map(7);
out.bulk(b"server");
out.bulk(REPORTED_SERVER.as_bytes());
out.bulk(b"version");
out.bulk(REPORTED_VERSION.as_bytes());
out.bulk(b"proto");
out.int(proto);
out.bulk(b"id");
out.int(session.id as i64);
out.bulk(b"mode");
out.bulk(b"standalone");
out.bulk(b"role");
out.bulk(b"master");
out.bulk(b"modules");
out.array(0);
Ok(())
}
fn command(args: Args<'_>, out: &mut Out) -> Result<()> {
if args.len() == 1 {
out.array(table::COMMANDS.len());
for spec in table::COMMANDS {
write_spec(out, spec);
}
return Ok(());
}
let sub = args.get(1);
if is(sub, b"COUNT") {
out.int(table::COMMANDS.len() as i64);
} else if is(sub, b"INFO") {
if args.len() == 2 {
out.array(table::COMMANDS.len());
for spec in table::COMMANDS {
write_spec(out, spec);
}
} else {
out.array(args.len() - 2);
for i in 2..args.len() {
match table::lookup(args.get(i)) {
Some(spec) => write_spec(out, spec),
None => out.nil(),
}
}
}
} else if is(sub, b"LIST") {
list(args, out)?;
} else if is(sub, b"DOCS") {
docs(args, out);
} else if is(sub, b"GETKEYS") {
getkeys(args, out)?;
} else if is(sub, b"HELP") {
help(out, COMMAND_HELP);
} else {
return Err(args::unknown_subcommand(sub, "COMMAND"));
}
Ok(())
}
fn list(args: Args<'_>, out: &mut Out) -> Result<()> {
if args.len() == 2 {
out.array(table::COMMANDS.len());
for spec in table::COMMANDS {
out.bulk(spec.name.as_bytes());
}
return Ok(());
}
if args.len() != 5 || !is(args.get(2), b"FILTERBY") {
return Err(args::syntax());
}
let (how, what) = (args.get(3), args.get(4));
let keep = |spec: &Spec| {
if is(how, b"MODULE") {
false
} else if is(how, b"ACLCAT") {
spec.acl
.iter()
.any(|c| c.len() == what.len() + 1 && c.as_bytes()[1..].eq_ignore_ascii_case(what))
} else {
glob::matches(what, spec.name.as_bytes())
}
};
if !is(how, b"MODULE") && !is(how, b"ACLCAT") && !is(how, b"PATTERN") {
return Err(args::syntax());
}
out.array(table::COMMANDS.iter().filter(|s| keep(s)).count());
for spec in table::COMMANDS.iter().filter(|s| keep(s)) {
out.bulk(spec.name.as_bytes());
}
Ok(())
}
fn docs(args: Args<'_>, out: &mut Out) {
if args.len() == 2 {
out.map(table::COMMANDS.len());
for spec in table::COMMANDS {
write_docs(out, spec);
}
return;
}
let found = (2..args.len())
.filter(|&i| table::lookup(args.get(i)).is_some())
.count();
out.map(found);
for i in 2..args.len() {
if let Some(spec) = table::lookup(args.get(i)) {
write_docs(out, spec);
}
}
}
fn write_docs(out: &mut Out, spec: &Spec) {
out.bulk(spec.name.as_bytes());
out.map(4);
out.bulk(b"summary");
out.bulk(spec.summary.as_bytes());
out.bulk(b"since");
out.bulk(spec.since.as_bytes());
out.bulk(b"group");
out.bulk(spec.group.as_bytes());
out.bulk(b"complexity");
out.bulk(spec.complexity.as_bytes());
}
fn getkeys(args: Args<'_>, out: &mut Out) -> Result<()> {
if args.len() < 3 {
return Err(args::wrong_arity_sub("command", "getkeys"));
}
let inner = args.get(2);
let spec = table::lookup(inner)
.ok_or_else(|| Error::new(Code::Unsupported, "Invalid command specified"))?;
let argc = args.len() - 2;
if !table::arity_ok(spec, argc) {
return Err(Error::new(
Code::Invalid,
"Invalid number of arguments specified for command",
));
}
if spec.name == "msetex" {
let n = parse_i64(args.get(3))
.filter(|&n| n > 0)
.and_then(|n| usize::try_from(n).ok())
.filter(|&n| 4 + 2 * n <= args.len())
.ok_or_else(|| Error::new(Code::Invalid, "Invalid arguments specified for command"))?;
out.array(n);
for i in 0..n {
out.bulk(args.get(4 + 2 * i));
}
return Ok(());
}
if spec.first_key == 0 {
return Err(Error::new(
Code::Invalid,
"The command has no key arguments",
));
}
let last = if spec.last_key < 0 {
(argc as i64) + i64::from(spec.last_key)
} else {
i64::from(spec.last_key)
};
let step = i64::from(spec.step).max(1);
let first = i64::from(spec.first_key);
let count = if last < first {
0
} else {
((last - first) / step + 1) as usize
};
out.array(count);
for i in 0..count {
out.bulk(args.get(2 + (first + (i as i64) * step) as usize));
}
Ok(())
}
fn write_spec(out: &mut Out, spec: &Spec) {
out.array(10);
out.bulk(spec.name.as_bytes());
out.int(i64::from(spec.arity));
out.array(spec.flags.len());
for f in spec.flags {
out.simple(f.as_bytes());
}
out.int(i64::from(spec.first_key));
out.int(i64::from(spec.last_key));
out.int(i64::from(spec.step));
out.array(spec.acl.len());
for a in spec.acl {
out.simple(a.as_bytes());
}
out.array(0);
out.array(0);
out.array(0);
}
fn read_knob(db: &Keyspace, knob: Knob) -> usize {
match knob {
Knob::SetIntsetEntries => db.limits().max_intset_entries,
Knob::SetListpackEntries => db.limits().max_listpack_entries,
Knob::SetListpackValue => db.limits().max_listpack_value,
Knob::HashListpackEntries => db.hash_limits().max_listpack_entries,
Knob::HashListpackValue => db.hash_limits().max_listpack_value,
}
}
fn write_knob(db: &mut Keyspace, knob: Knob, n: usize) {
let mut set = *db.limits();
let mut hash = *db.hash_limits();
match knob {
Knob::SetIntsetEntries => set.max_intset_entries = n,
Knob::SetListpackEntries => set.max_listpack_entries = n,
Knob::SetListpackValue => set.max_listpack_value = n,
Knob::HashListpackEntries => hash.max_listpack_entries = n,
Knob::HashListpackValue => hash.max_listpack_value = n,
}
db.set_limits(set);
db.set_hash_limits(hash);
}
fn bad_setting(name: &str, parsed: bool) -> Error {
if parsed {
Error::fmt(
Code::Invalid,
format_args!(
"CONFIG SET failed (possibly related to argument '{name}') - argument must be between 0 and 9223372036854775807 inclusive"
),
)
} else {
Error::fmt(
Code::Invalid,
format_args!(
"CONFIG SET failed (possibly related to argument '{name}') - argument couldn't be parsed into an integer"
),
)
}
}
fn config(server: &mut Server, args: Args<'_>, out: &mut Out) -> Result<()> {
let sub = args.get(1);
if is(sub, b"GET") {
if args.len() < 3 {
return Err(args::wrong_arity_sub("config", "get"));
}
let wanted =
|name: &str| (2..args.len()).any(|i| glob::matches(args.get(i), name.as_bytes()));
let fixed = SETTINGS.iter().filter(|(k, _)| wanted(k));
let ladder = LADDER.iter().filter(|(k, _)| wanted(k));
out.map(fixed.clone().count() + ladder.clone().count());
for (k, v) in fixed {
out.bulk(k.as_bytes());
out.bulk(v.as_bytes());
}
for (k, knob) in ladder {
out.bulk(k.as_bytes());
out.bulk_int(read_knob(server.db_ref(0), *knob) as i64);
}
} else if is(sub, b"SET") {
if args.len() < 4 {
return Err(args::wrong_arity_sub("config", "set"));
}
if !args.len().is_multiple_of(2) {
return Err(args::syntax());
}
let mut writes = [None; 8];
let mut count = 0;
let mut i = 2;
while i < args.len() {
let (name, value) = (args.get(i), args.get(i + 1));
i += 2;
if let Some((k, knob)) = LADDER.iter().find(|(k, _)| is(name, k.as_bytes())) {
let Some(n) = parse_i64(value).filter(|&n| n >= 0) else {
return Err(bad_setting(k, parse_i64(value).is_some()));
};
if count == writes.len() {
return Err(args::syntax());
}
writes[count] = Some((*knob, n as usize));
count += 1;
continue;
}
let Some((k, v)) = SETTINGS.iter().find(|(k, _)| is(name, k.as_bytes())) else {
return Err(yo_alloc::allow(|| {
Error::fmt(
Code::Invalid,
format_args!(
"Unknown option or number of arguments for CONFIG SET - '{}'",
String::from_utf8_lossy(name)
),
)
}));
};
if value != v.as_bytes() {
return Err(Error::fmt(
Code::Unsupported,
format_args!(
"CONFIG SET failed (possibly related to argument '{k}') - can't set immutable config"
),
));
}
}
for (knob, n) in writes.iter().flatten() {
for at in 0..DATABASES {
write_knob(server.db(at), *knob, *n);
}
}
out.ok();
} else if is(sub, b"RESETSTAT") {
server.stats.commands = 0;
server.stats.connections = 0;
out.ok();
} else if is(sub, b"REWRITE") {
return Err(Error::new(
Code::Unsupported,
"The server is running without a config file",
));
} else if is(sub, b"HELP") {
help(out, CONFIG_HELP);
} else {
return Err(args::unknown_subcommand(sub, "CONFIG"));
}
Ok(())
}
fn info(server: &Server, args: Args<'_>, out: &mut Out) {
let want = |section: &str| {
args.len() == 1
|| (1..args.len()).any(|i| {
let a = args.get(i);
is(a, section.as_bytes())
|| is(a, b"all")
|| is(a, b"everything")
|| is(a, b"default")
})
};
let text = yo_alloc::allow(|| {
let mut s = String::with_capacity(1024);
if want("server") {
let _ = write!(
s,
"# Server\r\nredis_version:{REPORTED_VERSION}\r\nyo_version:{}\r\n\
redis_mode:standalone\r\narch_bits:{}\r\nprocess_id:0\r\n\
run_id:0000000000000000000000000000000000000000\r\ntcp_port:0\r\n\
uptime_in_seconds:{}\r\nio_threads_active:0\r\n\r\n",
env!("CARGO_PKG_VERSION"),
usize::BITS,
server.uptime_secs(),
);
}
if want("clients") {
let _ = write!(
s,
"# Clients\r\nconnected_clients:{}\r\nblocked_clients:{}\r\n\
cluster_connections:0\r\n\r\n",
server.stats.clients,
server.waiters().len(),
);
}
if want("memory") {
let _ = write!(
s,
"# Memory\r\nused_memory:{}\r\nused_memory_dataset:{}\r\n\
used_memory_overhead:{}\r\nmem_arena_bytes:{}\r\n\
mem_arena_segments:{}\r\nmem_index_bytes:{}\r\n\
mem_client_buffers:{}\r\nmaxmemory:0\r\n\
maxmemory_policy:noeviction\r\n\r\n",
server.memory_bytes(),
server.dataset_bytes(),
server.memory_bytes() - server.dataset_bytes(),
server.arena_bytes(),
server.segment_count(),
server.index_bytes(),
server.conn_bytes(),
);
}
if want("stats") {
let _ = write!(
s,
"# Stats\r\ntotal_connections_received:{}\r\n\
total_commands_processed:{}\r\nexpired_keys:{}\r\n\r\n",
server.stats.connections,
server.stats.commands,
server.expired_keys(),
);
}
if want("cpu") {
if let Some(u) = cpu::usage() {
let _ = write!(
s,
"# CPU\r\nused_cpu_sys:{:.6}\r\nused_cpu_user:{:.6}\r\n\
used_cpu_sys_children:{:.6}\r\nused_cpu_user_children:{:.6}\r\n\r\n",
u.sys, u.user, u.sys_children, u.user_children,
);
}
}
if want("replication") {
s.push_str("# Replication\r\nrole:master\r\nconnected_slaves:0\r\n\r\n");
}
if want("keyspace") {
s.push_str("# Keyspace\r\n");
for i in 0..DATABASES {
let keys = server.dbs[i].len();
if keys > 0 {
let _ = write!(s, "db{i}:keys={keys},expires=0,avg_ttl=0\r\n");
}
}
s.push_str("\r\n");
}
s
});
out.verbatim(b"txt", text.as_bytes());
}
pub(super) fn help(out: &mut Out, lines: &[&str]) {
out.array(lines.len());
for line in lines {
out.simple(line.as_bytes());
}
}
const COMMAND_HELP: &[&str] = &[
"COMMAND <subcommand> [<arg> [value] [opt] ...]. Subcommands are:",
"(no subcommand)",
" Return details about all commands.",
"COUNT",
" Return the total number of commands in this server.",
"LIST [FILTERBY <MODULE <module-name>|ACLCAT <category>|PATTERN <pattern>>]",
" Return a list of all commands in this server.",
"INFO [<command-name> ...]",
" Return details about multiple commands.",
"DOCS [<command-name> ...]",
" Return documentation details about multiple commands.",
"GETKEYS <full-command>",
" Return the keys from a full command.",
"HELP",
" Print this help.",
];
const CONFIG_HELP: &[&str] = &[
"CONFIG <subcommand> [<arg> [value] [opt] ...]. Subcommands are:",
"GET <pattern>",
" Return parameters matching the glob-like <pattern> and their values.",
"SET <directive> <value>",
" Set the configuration <directive> to <value>.",
"RESETSTAT",
" Reset statistics reported by the INFO command.",
"REWRITE",
" Rewrite the configuration file.",
"HELP",
" Print this help.",
];