Lists
Vehicle headlights are on and they're not responding. Those shouldn't be running off the car batteries. Item 151 on today's glitch list.
In this chapter we will go through another extension of the CodeCrafters Redis Challenge: lists.
This is an interesting topic, as lists introduce two distinct problems. For starters, they are a new data type, and so far we only handled strings. We implemented commands that accept and return integers (e.g. INCR) but we handled the conversion to and from strings in the server command. Fortunately, back in chapter 3 we introduced StorageValue as a way to decouple the storage and the data type. In this chapter we will reap the fruits of this early concern. Adding the new type without that decoupling in place would have been possible, but would have required more work even in places where lists are not actually involved.
The second challenge presented by lists is that they are not a scalar type. Types like strings or integers are relatively easy to handle as they either exist in the storage or not. A string can be empty, but it is always treated like a single element. Conversely, array-like types can not only exist or not, but can be altered by adding and removing elements, and as we will see soon, such operations can be done at either end of the array.
If you exclude the refactoring needed to tackle the two problems mentioned above, the stages of the CodeCrafters challenge are relatively straightforward.
At the end of the chapter we are also going to implement the command BLPOP that requires handling blocking clients with timeouts. It will be an interesting problem to take, and it will allow us to appreciate once more the benefits of the actor model and to discuss resource ownership.
Step 10.1 - Create a list
The first stage in the challenge is simple to implement, but requires some changes to the storage. So far, we assumed the storage contained only strings, and the only time we handled a different type was with the command INCR. Even in that case, however, data was stored in string form and converted on the fly.
With the introduction of lists things change. Of course, lists can be transformed into strings and converted on the fly as well, but doing something like that seems unnecessarily complicated.
The introduction of StorageValue in chapter 3 already gives us all we need to host a different data type, but the method Storage::get is not ready to support that: currently, the signature of the method forces it to return StorageResult<Option<String>>, which is not exhaustive any more.
The plan for this step, therefore, is the following:
- Change the signature of
Storage::get and change the existing code to match the new implementation. This allows us to use get to fetch any type of data. - Add a new variant
StorageValue::List. This enables the system to store Redis lists. - Generalise how
StorageData values are built. Now that the storage can hold both strings and lists, we can replace From<String> with typed constructors. - Implement the command
RPUSH, as the challenge requires.
Step 10.1.1 - Change signature
Changing the signature of Storage::get is fairly simple.
src/storage.rs
impl Storage {
// Implement the `get` operation for the storage.
pub fn get(&mut self, key: &str) -> StorageResult<Option<String>> {
pub fn get(&mut self, key: &str) -> StorageResult<Option<StorageValue>> {
if let Some(&expiry) = self.expiry.get(key) {
if SystemTime::now() >= expiry {
self.expiry.remove(key);
self.store.remove(key);
return Ok(None);
}
}
match self.store.get(key) {
Some(StorageData {
value: StorageValue::String(v),
creation_time: _,
expiry: _,
}) => Ok(Some(v.clone())),
}) => Ok(Some(StorageValue::String(v.clone()))),
None => Ok(None),
}
}
}This has an immediate effect on tests and commands that use the method. Let's change the tests first.
src/storage.rs
mod tests {
#[test]
// Test that the function get works as expected.
// When a key value is retrieved, the output
// is the value, and the key is not deleted
// from the storage.
fn test_get_value() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::from(String::from("avalue")),
);
let result = storage.get("akey").unwrap();
assert_eq!(storage.store.len(), 1);
assert_eq!(result, Some(String::from("avalue")));
assert_eq!(result, Some(StorageValue::String(String::from("avalue"))));
}The different signature requires changes to the command GET.
src/commands/get.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
use crate::storage::StorageValue;
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
match output {
Ok(Some(v)) => request.data(ServerValue::Resp(Resp::BulkString(v))).await,
Ok(Some(StorageValue::String(v))) => {
request.data(ServerValue::Resp(Resp::BulkString(v))).await
}
Ok(None) => request.data(ServerValue::Resp(Resp::Null)).await,
Err(_) => {
request
.error(ServerError::CommandInternalError(command.join(" ")))
.await
}
};The command INCR needs similar changes.
src/commands/incr.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
use crate::set::SetArgs;
use crate::storage::StorageValue;
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
Ok(storage_value) => {
// Convert None into a "0".
let storage_value = storage_value.unwrap_or_else(|| String::from("0"));
let storage_value = match storage_value {
Some(StorageValue::String(s)) => s,
None => String::from("0"),
};
mod tests {
async fn test_command_key_exists() {
assert_eq!(
connection_receiver.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::Integer(43)))
);
assert_eq!(
server.storage.as_mut().unwrap().get("key").unwrap(),
Some("43".to_string())
Some(StorageValue::String(String::from("43")))
);Step 10.1.2 - Add new variant
We can now add a new variant to StorageValue. In a compiled language like Rust, adding a new variant forces us to update code that matches on the enum, since every match now has a new case to cover (unless a catch-all branch already exists).
src/storage.rs
#[derive(Debug, PartialEq)]
pub enum StorageValue {
String(String),
List(Vec<String>),
}
impl Storage {
pub fn get(&mut self, key: &str) -> StorageResult<Option<StorageValue>> {
match self.store.get(key) {
Some(StorageData {
value: StorageValue::String(v),
creation_time: _,
expiry: _,
}) => Ok(Some(StorageValue::String(v.clone()))),
None => Ok(None),
_ => Ok(None),
}Here, we transformed the specific branch None => into a generic catch-all _ => to cover the additional case without having to implement logic for it. We will revisit this piece of code later, as the addition of commands that read lists will generate specific cases that we want to handle. For the time being, however, this code implements the correct behaviour.
There is another match that needs to be adjusted, in the implementation of the server command GET.
src/commands/get.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
match output {
Ok(Some(StorageValue::String(v))) => {
request.data(ServerValue::Resp(Resp::BulkString(v))).await
}
Ok(None) => request.data(ServerValue::Resp(Resp::Null)).await,
Err(_) => {
_ => {
request
.error(ServerError::CommandInternalError(command.join(" ")))
.await
}
};Once again, this is the correct solution for the time being, but we will soon restore the more specific case Err(_) => when the code will handle lists.
The most relevant change, in terms of amount of code, occurs in the implementation of the command INCR, where we need to check if the value is a string or not, as there is no implementation of this command for lists. The change is trivial in theory, but the function already contains a noticeable amount of nesting to handle several error cases, so it's worth rewriting it completely, splitting the logic into two steps: fetch the value from the storage and check its type first, then handle parsing and the increment operation.
Given the extent of the change, the full code of the function will be shown to highlight exactly which parts have been changed.
src/commands/incr.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Extract the key from the command line.
let key = &command[1];
// Find the value of the key.
let output = storage.get(key);
// Check the value of the key.
let result = match output {
...
};
// Resolve the stored string. The key might exist or not,
// and if it exists it might not be a StorageValue::String.
let string_value = match output {
// The key exists and holds a string.
Ok(Some(StorageValue::String(s))) => s,
// The key doesn't exist. Pretend it is "0".
Ok(None) => String::from("0"),
// The key exists but isn't a string, or reading it failed.
Ok(Some(_)) | Err(_) => {
request
.error(ServerError::CommandInternalError(command.join(" ")))
.await;
return;
}
};
// Parse the string value into an i64.
let result = match string_value.parse::<i64>() {
// The conversion succeeded.
Ok(n) => {
// Increment the value.
let value = n + 1;
// Try to store the incremented value.
match storage.set(key.clone(), value.to_string(), SetArgs::new()) {
// Incremented value successfully stored,
// send it back to the client.
Ok(_) => Ok(ServerValue::Resp(Resp::Integer(value))),
// An error occurred.
Err(_) => Err(ServerError::CommandInternalError(command.join(" "))),
}
}
// The value is not an integer.
Err(_) => Ok(ServerValue::Resp(Resp::simple_error(
"ERR value is not an integer or out of range",
))),
};
request.result(result).await;When we apply this type of change, we can take full advantage of the existing tests. The tests were passing before we added the new variant, so we can delete the code completely and rebuild it from scratch until we reach a state where the test suite gives us the green light again.
Step 10.1.3 - Change the storage
Before we implement the command RPUSH, we need to make sure the storage is configured to handle non-string data. While StorageData is connected with StorageValue, and thus automatically ready to accept new variants, the implementation of its method From<String> is not. The code has the same issues we discussed for Resp in the previous chapter, as the conversion between the two types is not strictly defined.
Let's start with a more generic implementation of From<String>.
src/storage.rs
impl From<String> for StorageData {
fn from(s: String) -> Self {
impl From<StorageValue> for StorageData {
fn from(value: StorageValue) -> Self {
Self {
value: StorageValue::String(s),
value,
creation_time: SystemTime::now(),
expiry: None,
}
}
}This can be complemented by two helpers to build items that contain either StorageValue::String or StorageValue::List.
src/storage.rs
impl StorageData {
pub fn add_expiry(&mut self, expiry: Duration) {
self.expiry = Some(expiry);
}
pub fn string(s: impl Into<String>) -> Self {
Self::from(StorageValue::String(s.into()))
}
pub fn list<I, T>(items: I) -> Self
where
I: IntoIterator<Item = T>,
T: Into<String>,
{
Self::from(StorageValue::List(
items.into_iter().map(Into::into).collect(),
))
}
}As mentioned before, the solution is extremely similar to what we have done with Resp in the previous chapter. The only noticeable difference is that StorageData::list uses map(Into::into) to call .into() on each element, whereas Resp::bulk_array uses a helper function.
The change affects the implementation of Storage::set.
src/storage.rs
impl Storage {
// Implement the `set` operation for the storage.
pub fn set(&mut self, key: String, value: String, args: SetArgs) -> StorageResult<String> {
let mut data = StorageData::from(value);
let mut data = StorageData::string(value);
if let Some(value) = args.expiry {
let expiry = match value {
KeyExpiry::EX(v) => Duration::from_secs(v),
KeyExpiry::PX(v) => Duration::from_millis(v),
};
data.add_expiry(expiry);
self.expiry
.insert(key.clone(), data.creation_time.add(expiry));
}
self.store.insert(key, data);
Ok(String::from("OK"))
}It will also affect some of the tests written for Storage.
src/storage.rs
mod tests {
fn test_set_value() {
let mut storage: Storage = Storage::new();
let avalue = StorageData::from(String::from("avalue"));
let avalue = StorageData::string("avalue");
let output = storage
.set(String::from("akey"), String::from("avalue"), SetArgs::new())
.unwrap();
fn test_set_value_with_px() {
let mut storage: Storage = Storage::new();
let mut avalue = StorageData::from(String::from("avalue"));
let mut avalue = StorageData::string("avalue");
avalue.add_expiry(Duration::from_millis(100));
fn test_get_value() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::from(String::from("avalue")),
);
storage
.store
.insert(String::from("akey"), StorageData::string("avalue"));
let result = storage.get("akey").unwrap();Being an implementation detail, StorageData is not used outside src/storage.rs, so there are no changes to other modules.
Step 10.1.4 - Implement RPUSH
As with GET and SET, the command RPUSH has two sides. One is the internal implementation in Storage, the other is the external (or client-facing) implementation of the server command.
Thanks to this decoupling, the internal implementation can exist independently of the external one, which allows us to implement and test them in stages. Let's tackle the internal implementation first.
The function needs to get the requested key from the storage first. The resulting value has to be mutable as we want to add an item to the list.
src/storage.rs
use crate::set::{KeyExpiry, SetArgs};
use crate::storage_result::StorageResult;
use crate::storage_result::{StorageError, StorageResult};
use std::collections::HashMap;
use std::ops::Add;
use std::time::{Duration, SystemTime};
impl Storage {
// Implement the `rpush` operation for the storage.
pub fn rpush(&mut self, key: &str, value: String) -> StorageResult<usize> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let length = match self.store.get_mut(key) {
// CODE HERE
};
Ok(length)
}The core of the function is a match that handles three cases. The first one is "the key doesn't exist". This is currently implemented with a catch-all _ that will be later converted into None.
src/storage.rs
impl Storage {
// Implement the `rpush` operation for the storage.
pub fn rpush(&mut self, key: &str, value: String) -> StorageResult<usize> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let length = match self.store.get_mut(key) {
// The key doesn't exist. Create it as
// a list of a single element and
// return the length of 1.
_ => {
self.store
.insert(key.to_string(), StorageData::list([value]));
1
}
};
Ok(length)
}
mod tests {
#[test]
// Test that the function rpush works as expected
// when the key doesn't exist.
// The key is created as a list and the value is
// added to it.
fn test_rpush_key_does_not_exist() {
let mut storage: Storage = Storage::new();
let avalue = StorageData::list(["avalue"]);
let output = storage.rpush("akey", String::from("avalue")).unwrap();
assert_eq!(output, 1);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &avalue),
None => panic!(),
}
}The second case is "the key exists but the value is a string".
src/storage.rs
impl Storage {
// Implement the `rpush` operation for the storage.
pub fn rpush(&mut self, key: &str, value: String) -> StorageResult<usize> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let length = match self.store.get_mut(key) {
// The value is a string, we cannot
// use rpush on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => return Err(StorageError::WrongType),
// The key doesn't exist. Create it as
// a list of a single element and
// return the length of 1.
_ => {
self.store
.insert(key.to_string(), StorageData::list([value]));
1
}
};
Ok(length)
}
mod tests {
#[test]
// Test that the function rpush
// returns the correct error when
// the key exists and is not a list.
fn test_rpush_key_exists_and_is_not_list() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::string("avalue"));
let error = storage.rpush("akey", String::from("value2"));
assert_eq!(error, Err(StorageError::WrongType));
}The new error StorageError::WrongType used in the function is a simple variant.
src/storage_result.rs
#[derive(Debug, PartialEq)]
pub enum StorageError {
CommandSyntaxError(String),
WrongType,
}
impl fmt::Display for StorageError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::CommandSyntaxError(string) => {
write!(f, "Syntax error while processing {}!", string)
}
Self::WrongType => {
write!(f, "Wrong type")
}
}
}
}The third and last case is "the key exists and the value is a list".
src/storage.rs
impl Storage {
// Implement the `rpush` operation for the storage.
pub fn rpush(&mut self, key: &str, value: String) -> StorageResult<usize> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let length = match self.store.get_mut(key) {
// The value is a string, we cannot
// use rpush on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => return Err(StorageError::WrongType),
// The value is already a list. Add
// the new element and return the
// length.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
v.push(value);
v.len()
}
// The key doesn't exist. Create it as
// a list of a single element and
// return the length of 1.
_ => {
self.store
.insert(key.to_string(), StorageData::list([value]));
1
}
};
Ok(length)
}
mod tests {
#[test]
// Test that the function rpush works as expected
// when the key exists and is a list.
// The value is added to it.
fn test_rpush_key_exists_and_is_list() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::list(["value1"]));
let list = StorageData::list(["value1", "value2"]);
let output = storage.rpush("akey", String::from("value2")).unwrap();
assert_eq!(output, 2);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &list),
None => panic!(),
}
}At this point, the catch-all _ introduced at the beginning can be transformed into a more specific None.
src/storage.rs
impl Storage {
// Implement the `rpush` operation for the storage.
pub fn rpush(&mut self, key: &str, value: String) -> StorageResult<usize> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let length = match self.store.get_mut(key) {
// The value is a string, we cannot
// use rpush on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => return Err(StorageError::WrongType),
// The value is already a list. Add
// the new element and return the
// length.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
v.push(value);
v.len()
}
// The key doesn't exist. Create it as
// a list of a single element and
// return the length of 1.
_ => {
None => {
self.store
.insert(key.to_string(), StorageData::list([value]));
1
}
};
Ok(length)
}With this code in place, we can expose the function in the server. As you might have already noticed, all functions exposing server commands are extremely similar: they might have to prepare the server (or the storage contained in it), check or transform the parameters, and finally call some logic and handle its output. The implementation of RPUSH is not different.
src/commands/rpush.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Extract the storage from the server.
let storage = match server.storage.as_mut() {
Some(storage) => storage,
None => {
request.error(ServerError::StorageNotInitialised).await;
return;
}
};
// Check that the command received a key and a value.
if command.len() != 3 {
request
.error(ServerError::CommandSyntaxError(command.join(" ")))
.await;
return;
}
// Extract the key and the value.
let key = &command[1];
let value = command[2].clone();
let result = storage.rpush(key, value);
match result {
Ok(size) => {
request
.data(ServerValue::Resp(Resp::Integer(size as i64)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}
}As usual, the function can be tested. The first use case is when the command is called with a single value.
src/commands/rpush.rs
#[cfg(test)]
mod tests {
use super::*;
use crate::server_result::ServerMessage;
use crate::storage::Storage;
use tokio::sync::mpsc;
#[tokio::test]
// Test that the function command processes
// an `RPUSH` request with a single value.
async fn test_command_single_value() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("rpush"),
String::from("key"),
String::from("value"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::Integer(1)))
);
}
}We then need to check the output of the function when the server hasn't been initialised with some storage.
src/commands/rpush.rs
mod tests {
#[tokio::test]
// Test that the function command returns the
// correct error when called on a server that
// has no storage attached.
async fn test_command_no_storage() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let cmd = vec![
String::from("rpush"),
String::from("key"),
String::from("value"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::StorageNotInitialised)
);
}Last, we need to check the function's behaviour when we pass the wrong number of parameters.
src/commands/rpush.rs
mod tests {
#[tokio::test]
// Test that the function command returns the
// correct error when called with the wrong
// number of parameters.
async fn test_command_wrong_number_of_parameters() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![String::from("rpush"), String::from("key")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::CommandSyntaxError(String::from("rpush key")))
);
}The new module must be exposed so that the server can import it.
src/commands/mod.rs
pub mod discard;
pub mod echo;
pub mod exec;
pub mod get;
pub mod incr;
pub mod info;
pub mod multi;
pub mod ping;
pub mod psync;
pub mod replconf;
pub mod rpush;
pub mod set;
pub mod wait;
Last, we need to expose the command in the server interface.
src/server.rs
use crate::client::Client;
use crate::commands::{
discard, echo, exec, get, incr, info, multi, ping, psync, replconf, set, wait,
discard, echo, exec, get, incr, info, multi, ping, psync, replconf, rpush, set, wait,
};
use crate::connection::{
stream_read_data_length, stream_read_line, stream_send_receive_resp, ConnectionMessage,
};
use crate::replication::ReplicationConfig;
use crate::request::Request;
use crate::resp::{bytes_to_resp, resp_extract_length, resp_remove_type};
use crate::server_result::{ServerError, ServerMessage, ServerResult, ServerValue};
use crate::storage::Storage;
use crate::Resp;
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::{io::AsyncWriteExt, net::TcpStream};
pub async fn process_request(request: Request, server: &mut Server) {
match command_name.as_str() {
"replconf" => {
replconf::command(server, &request, &command).await;
}
"rpush" => {
rpush::command(server, &request, &command).await;
}
"set" => {
set::command(server, &request, &command).await;
// Forward the request to all replicas.
send_request_to_replicas(&request, server).await;
}CodeCrafters
Lists Stage 1: Create a list
The code we wrote in this section passes Lists - Stage 1 of the CodeCrafters challenge.
Step 10.2 - Append an element
In this step, we need to extend RPUSH to append elements to existing lists. The code we wrote so far, however, has already been designed to do that. The reason is that we designed StorageValue::List to be an actual array (Vec<String>).
In this case, it is more difficult to design a system that works only with an empty list first than to implement what passes both stages at the same time. We can however make the two requirements explicit by splitting the test test_rpush_key_exists_and_is_list in two different tests, one with an empty list and one with a list that already contains some elements.
src/storage.rs
mod tests {
#[test]
// Test that the function rpush works as expected
// when the key exists and is a list.
// The value is added to it.
fn test_rpush_key_exists_and_is_list() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::list(["value1"]));
let list = StorageData::list(["value1", "value2"]);
let output = storage.rpush("akey", String::from("value2")).unwrap();
assert_eq!(output, 2);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &list),
None => panic!(),
}
}
#[test]
// Test that the function rpush works as expected
// when the key exists and is an empty list.
// The value is added to it.
fn test_rpush_key_exists_and_is_list_empty() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(Vec::<String>::new()),
);
let list = StorageData::list(["value1"]);
let output = storage.rpush("akey", String::from("value1")).unwrap();
assert_eq!(output, 1);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &list),
None => panic!(),
}
}
#[test]
// Test that the function rpush works as expected
// when the key exists and is a non-empty list.
// The value is added to it.
fn test_rpush_key_exists_and_is_list_not_empty() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::list(["value1"]));
let list = StorageData::list(["value1", "value2"]);
let output = storage.rpush("akey", String::from("value2")).unwrap();
assert_eq!(output, 2);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &list),
None => panic!(),
}
}CodeCrafters
Lists Stage 2: Append an element
The code we wrote in this section passes Lists - Stage 2 of the CodeCrafters challenge.
Step 10.3 - Append multiple elements
The next step naturally extends the command RPUSH to accept multiple elements and to append them all to the list. The new feature clearly requires a change of the signature and code of Storage::rpush, but overall remains quite contained.
src/storage.rs
impl Storage {
// Implement the `rpush` operation for the storage.
pub fn rpush(&mut self, key: &str, value: String) -> StorageResult<usize> {
pub fn rpush(&mut self, key: &str, values: &[String]) -> StorageResult<usize> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let length = match self.store.get_mut(key) {
// The value is a string, we cannot
// use rpush on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => return Err(StorageError::WrongType),
// The value is already a list. Add
// the new element and return the
// length.
// length of the whole list.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
v.push(value);
// Create an iterator of &String and own each
// element cloning it into a String.
v.extend(values.iter().cloned());
v.len()
}
// The key doesn't exist. Create it as
// a list of a single element and
// return the length of 1.
// return its length.
None => {
self.store
.insert(key.to_string(), StorageData::list([value]));
1
.insert(key.to_string(), StorageData::list(values.to_vec()));
values.len()
}
};
Ok(length)
}These changes affect the tests of the function.
src/storage.rs
mod tests {
fn test_rpush_key_does_not_exist() {
let mut storage: Storage = Storage::new();
let avalue = StorageData::list(["avalue"]);
let output = storage.rpush("akey", String::from("avalue")).unwrap();
let output = storage.rpush("akey", &[String::from("avalue")]).unwrap();
fn test_rpush_key_exists_and_is_list_empty() {
let list = StorageData::list(["value1"]);
let output = storage.rpush("akey", String::from("value1")).unwrap();
let output = storage.rpush("akey", &[String::from("value1")]).unwrap();
fn test_rpush_key_exists_and_is_list_not_empty() {
let list = StorageData::list(["value1", "value2"]);
let output = storage.rpush("akey", String::from("value2")).unwrap();
let output = storage.rpush("akey", &[String::from("value2")]).unwrap();
fn test_rpush_key_exists_and_is_not_list() {
storage
.store
.insert(String::from("akey"), StorageData::string("avalue"));
let error = storage.rpush("akey", String::from("value2"));
let error = storage.rpush("akey", &[String::from("value2")]);We can also add a new test to check that multiple values are handled correctly.
src/storage.rs
mod tests {
#[test]
// Test that the function rpush works as expected
// when the key exists and multiple values are
// pushed.
fn test_rpush_multiple_elements() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::list(["value1"]));
let list = StorageData::list(["value1", "value2", "value3"]);
let output = storage
.rpush("akey", &[String::from("value2"), String::from("value3")])
.unwrap();
assert_eq!(output, 3);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &list),
None => panic!(),
}
}Last, the changes propagate to the implementation of the command RPUSH.
src/commands/rpush.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Extract the storage from the server.
let storage = match server.storage.as_mut() {
Some(storage) => storage,
None => {
request.error(ServerError::StorageNotInitialised).await;
return;
}
};
// Check that the command received a key and a value.
if command.len() != 3 {
if command.len() < 3 {
request
.error(ServerError::CommandSyntaxError(command.join(" ")))
.await;
return;
}
// Extract the key and the value.
// Extract the key and the values.
let key = &command[1];
let value = command[2].clone();
let value = &command[2..];
let result = storage.rpush(key, value);
match result {
Ok(size) => {
request
.data(ServerValue::Resp(Resp::Integer(size as i64)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}
}All existing tests of the server command pass, as the command still supports the previous version of the syntax, but it's worth adding a new test to check that multiple arguments are accepted.
src/commands/rpush.rs
mod tests {
#[tokio::test]
// Test that the function command processes
// a `RPUSH` request with multiple values
// to an existing key which is already a list.
async fn test_command_multiple_values() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("rpush"),
String::from("key"),
String::from("value0"),
String::from("value1"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::Integer(2)))
);
}CodeCrafters
Lists Stage 3: Append multiple elements
The code we wrote in this section passes Lists - Stage 3 of the CodeCrafters challenge.
Step 10.4 - List elements (positive indexes)
In the next step we are tasked to implement the simplest version of LRANGE, with only positive indexes. Even with this restriction, the implementation of such a command is not trivial, since there are several specific behaviours to keep in mind, as listed in the CodeCrafters challenge:
- If the list doesn't exist, an empty array is returned.
- If the start index is greater than or equal to the list's length, an empty array is returned.
- If the stop index is greater than or equal to the list's length, the stop index is treated as the last element.
- If the start index is greater than the stop index, an empty array is returned.
Therefore, we will add many tests for the internal implementation Storage::lrange. This in turn will highlight that the implementation of Storage contains too many functions (and, by extension, too many tests), which will lead to a small refactoring in the next step before we move on with the challenge.
For now, let's focus on Storage::lrange. The overall structure is externally similar to that of Storage::rpush, since the function has to handle the same error cases when the key corresponds to a string or when the key doesn't exist.
src/storage.rs
impl Storage {
// Implement the `lrange` operation for the storage.
pub fn lrange(&self, key: &str, start: i64, stop: i64) -> StorageResult<&[String]> {
match self.store.get(key) {
// The value is a string, we cannot
// use lrange on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => Err(StorageError::WrongType),
// The value is a list. All good,
// extract the elements and
// return them.
Some(StorageData {
value: StorageValue::List(v),
..
}) => Ok(v),
// The key doesn't exist.
// Return an empty array.
None => Ok(&[]),
}
}The happy path, where the key exists and is a list, has been temporarily stubbed to return the whole list, to give us time to write tests for the error cases.
src/storage.rs
mod tests {
#[test]
// Test that the function lrange works as expected
// when the key doesn't exist.
fn test_lrange_empty_list() {
let storage: Storage = Storage::new();
let expected = Vec::<String>::new();
let output = storage.lrange("akey", 1, 3).unwrap();
assert_eq!(output, expected);
}
#[test]
// Test that the function lrange
// returns the correct error when
// the key exists and is not a list.
fn test_lrange_key_exists_and_is_not_list() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::string("avalue"));
let error = storage.lrange("akey", 1, 3);
assert_eq!(error, Err(StorageError::WrongType));
}With this in place, we can start implementing the required behaviours. Assuming everything is correct, start and stop will be passed from outside and will be within the boundaries of the list, so we can transform them into usize (needed to index a Vec) and return the resulting slice.
Now that we have a working structure we can follow a stricter TDD approach. This is the test for the happy path.
src/storage.rs
mod tests {
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list, the indexes
// are valid, and start is less than stop.
fn test_lrange() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = vec![
String::from("value1"),
String::from("value2"),
String::from("value3"),
];
let output = storage.lrange("akey", 1, 3).unwrap();
assert_eq!(output, expected);
}And the code that passes it is the following:
src/storage.rs
impl Storage {
pub fn lrange(&self, key: &str, start: i64, stop: i64) -> StorageResult<&[String]> {
// The value is a list. All good,
// extract the elements and
// return them.
Some(StorageData {
value: StorageValue::List(v),
..
}) => Ok(v),
}) => {
// Go back to usize to index
// the Rust vector.
let start = start as usize;
let stop = stop as usize;
Ok(&v[start..=stop])
}Now it's time to implement the advanced behaviours. The following two tests check the case when the start index is greater or equal to the length of the list.
src/storage.rs
mod tests {
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// and start > list length.
fn test_lrange_start_greater_than_length() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = Vec::<String>::new();
let output = storage.lrange("akey", 6, 7).unwrap();
assert_eq!(output, expected);
}
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// and start == list length.
fn test_lrange_start_equal_length() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = Vec::<String>::new();
let output = storage.lrange("akey", 5, 7).unwrap();
assert_eq!(output, expected);
}The following test checks the case when start is greater than stop.
src/storage.rs
mod tests {
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// and start > stop.
fn test_lrange_start_greater_than_stop() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = Vec::<String>::new();
let output = storage.lrange("akey", 3, 2).unwrap();
assert_eq!(output, expected);
}Strict TDD
Once again, remember that the book is not showing every step of a strict TDD approach. The TDD best practice is to add one test at a time and to make sure code passes that before moving on. Here, we see the final output of this process for multiple tests that check different edge cases of the same feature.
Here is the code that passes the tests:
src/storage.rs
impl Storage {
pub fn lrange(&self, key: &str, start: i64, stop: i64) -> StorageResult<&[String]> {
// The value is a list. All good,
// extract the elements and
// return them.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
// Make sure everything is i64.
let length = v.len() as i64;
// If start is past the end of
// the list or if start is greater
// than stop return an empty
// vector.
if start >= length || start > stop {
return Ok(&[]);
}
// Go back to usize to index
// the Rust vector.
let start = start as usize;
let stop = stop as usize;
Ok(&v[start..=stop])
}Last, we need to implement the behaviour of stop. When the index is greater or equal to the length of the list, the actual index becomes that of the last element.
src/storage.rs
mod tests {
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// and stop > list length.
fn test_lrange_stop_greater_than_length() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = vec![String::from("value3"), String::from("value4")];
let output = storage.lrange("akey", 3, 7).unwrap();
assert_eq!(output, expected);
}
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// and stop == list length.
fn test_lrange_stop_equal_length() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = vec![String::from("value3"), String::from("value4")];
let output = storage.lrange("akey", 3, 5).unwrap();
assert_eq!(output, expected);
}Passing the tests is straightforward:
src/storage.rs
use crate::set::{KeyExpiry, SetArgs};
use crate::storage_result::{StorageError, StorageResult};
use std::cmp::min;
use std::collections::HashMap;
use std::ops::Add;
use std::time::{Duration, SystemTime};
impl Storage {
pub fn lrange(&self, key: &str, start: i64, stop: i64) -> StorageResult<&[String]> {
// The value is a list. All good,
// extract the elements and
// return them.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
// Make sure everything is i64.
let length = v.len() as i64;
// If start is past the end of
// the list or if start is greater
// than stop return an empty
// vector.
if start >= length || start > stop {
return Ok(&[]);
}
// If stop is greater than
// the length of the list,
// replace it with the
// index of the last element.
let stop = min(stop, length - 1);
// Go back to usize to index
// the Rust vector.
let start = start as usize;
let stop = stop as usize;
Ok(&v[start..=stop])
}There is a potential bug that is worth exploring here. We check the condition start > stop, but then we modify stop if the value is greater than the length of the list.
So, start can be greater than stop either because the original input provides such values (e.g. 12, 10) or because start ends up being greater than stop after the latter is reduced to match the length of the list.
If you think about it, in either case start is greater than the length of the list, which leads to the same result (empty output), but it is worth checking the two cases explicitly.
src/storage.rs
mod tests {
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// start > stop, and stop > list length.
// Start is greater than stop BEFORE this is
// checked against the length of the list.
fn test_lrange_start_greater_than_stop_before_threshold() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = Vec::<String>::new();
let output = storage.lrange("akey", 12, 10).unwrap();
assert_eq!(output, expected);
}
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// and stop > list length
// Start is greater than stop AFTER this is
// checked against the length of the list.
fn test_lrange_start_greater_than_stop_after_threshold() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = Vec::<String>::new();
let output = storage.lrange("akey", 12, 15).unwrap();
assert_eq!(output, expected);
}Now that the implementation of Storage::lrange is completed, we can expose the command in the server. As usual, we first create and test the command function.
The function is similar to the implementation of RPUSH and other commands, with the usual tests to check the missing storage and the number of arguments.
src/commands/lrange.rs
#[cfg(test)]
mod tests {
use super::*;
use crate::server_result::ServerMessage;
use crate::storage::Storage;
use tokio::sync::mpsc;
#[tokio::test]
// Test that the function command processes
// an `LRANGE` request.
async fn test_command() {
let mut storage = Storage::new();
let _ = storage.rpush(
"key",
&[
String::from("value0"),
String::from("value1"),
String::from("value2"),
String::from("value3"),
String::from("value4"),
],
);
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("lrange"),
String::from("key"),
String::from("1"),
String::from("3"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::bulk_array(vec![
"value1", "value2", "value3"
])))
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called on a server that
// has no storage attached.
async fn test_command_no_storage() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let cmd = vec![
String::from("lrange"),
String::from("key"),
String::from("1"),
String::from("2"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::StorageNotInitialised)
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called with the wrong
// number of arguments.
async fn test_command_wrong_arguments() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("lrange"),
String::from("key"),
String::from("1"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::CommandSyntaxError(String::from(
"lrange key 1"
)))
);
}
}There are, however, two additional calls to test and implement, as the parameters start and stop passed on the command line must be converted from strings to numbers.
src/commands/lrange.rs
mod tests {
#[tokio::test]
// Test that the function command returns the
// correct error when the start index is
// not a number.
async fn test_command_start_index_not_number() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("lrange"),
String::from("key"),
String::from("apple"),
String::from("4"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::simple_error(
"ERR start value is not an integer or out of range"
)))
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when the stop index is
// not a number.
async fn test_command_stop_index_not_number() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("lrange"),
String::from("key"),
String::from("4"),
String::from("apple"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::simple_error(
"ERR stop value is not an integer or out of range"
)))
);
}The code that implements the server command is the following:
src/commands/lrange.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Extract the storage from the server.
let storage = match server.storage.as_mut() {
Some(storage) => storage,
None => {
request.error(ServerError::StorageNotInitialised).await;
return;
}
};
// Check that the command received 3 arguments.
if command.len() != 4 {
request
.error(ServerError::CommandSyntaxError(command.join(" ")))
.await;
return;
}
// Extract the key.
let key = &command[1];
// Extract start index.
let start: i64 = match command[2].parse() {
Ok(n) => n,
// The value is not an integer.
Err(_) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"ERR start value is not an integer or out of range",
)))
.await;
return;
}
};
// Extract stop index.
let stop: i64 = match command[3].parse() {
Ok(n) => n,
// The value is not an integer.
Err(_) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"ERR stop value is not an integer or out of range",
)))
.await;
return;
}
};
let result = storage.lrange(key, start, stop);
match result {
Ok(v) => {
let resp: Vec<Resp> = v.iter().cloned().map(Resp::BulkString).collect();
request.data(ServerValue::Resp(Resp::Array(resp))).await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}
}The last set of changes adds the command to the server.
src/commands/mod.rs
pub mod discard;
pub mod echo;
pub mod exec;
pub mod get;
pub mod incr;
pub mod info;
pub mod lrange;
pub mod multi;
pub mod ping;
pub mod psync;
pub mod replconf;
pub mod rpush;
pub mod set;
pub mod wait;
src/server.rs
use crate::client::Client;
use crate::commands::{
discard, echo, exec, get, incr, info, multi, ping, psync, replconf, rpush, set, wait,
discard, echo, exec, get, incr, info, lrange, multi, ping, psync, replconf, rpush, set, wait,
};
use crate::connection::{
stream_read_data_length, stream_read_line, stream_send_receive_resp, ConnectionMessage,
};
pub async fn process_request(request: Request, server: &mut Server) {
"info" => {
info::command(server, &request, &command).await;
}
"lrange" => {
lrange::command(server, &request, &command).await;
}
"ping" => {
ping::command(server, &request, &command).await;
}CodeCrafters
Lists Stage 4: List elements (positive indexes)
The code we wrote in this section passes Lists - Stage 4 of the CodeCrafters challenge.
Step 10.5 - List elements (negative indexes)
Once again, the new step adds several requirements to the logic behind LRANGE. When indexes are negative, we have the following rules:
- A negative index is an offset from the end of the list: -1 refers to the last element, -2 refers to the second-to-last, and so on.
- If a negative index is out of range (e.g., -6 on a list of length 5), it should be treated as 0 (the start of the list).
The good news is that these new rules only require an early check of the indexes and an eventual transformation.
The tests extracted from those requirements are the following:
src/storage.rs
mod tests {
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// and both start and stop are negative,
// but within the boundaries.
fn test_lrange_start_stop_negative() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = vec![
String::from("value1"),
String::from("value2"),
String::from("value3"),
];
let output = storage.lrange("akey", -4, -2).unwrap();
assert_eq!(output, expected);
}
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// both start and stop are negative,
// and start is out of range.
fn test_lrange_start_negative_out_of_range() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = vec![
String::from("value0"),
String::from("value1"),
String::from("value2"),
String::from("value3"),
];
let output = storage.lrange("akey", -100, -2).unwrap();
assert_eq!(output, expected);
}
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// both start and stop are negative,
// and stop is out of range.
fn test_lrange_stop_negative_out_of_range() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = Vec::<String>::new();
let output = storage.lrange("akey", -2, -100).unwrap();
assert_eq!(output, expected);
}
#[test]
// Test that the function lrange works as expected
// when the key exists and is a list,
// both start and stop are negative,
// and both are out of range.
fn test_lrange_start_stop_negative_out_of_range() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2", "value3", "value4"]),
);
let expected = Vec::<String>::new();
let output = storage.lrange("akey", -200, -100).unwrap();
assert_eq!(output, expected);
}And the changes to the code of Storage::lrange are the following:
src/storage.rs
impl Storage {
pub fn lrange(&self, key: &str, start: i64, stop: i64) -> StorageResult<&[String]> {
// The value is a list. All good,
// extract the elements and
// return them.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
// Make sure everything is i64.
let length = v.len() as i64;
// If start is negative add the
// length of the vector to make it
// relative to the end. Also, make
// sure 0 is the minimum value.
let start = if start < 0 {
(start + length).max(0)
} else {
start
};
// If stop is negative add the
// length of the vector to make it
// relative to the end.
// If the result is still negative
// keep it as it is. This will
// make start > stop and trigger
// the next check.
let stop = if stop < 0 { stop + length } else { stop };
// If start is past the end of
// the list or if start is greater
// than stop return an empty
// vector.
if start >= length || start > stop {
return Ok(&[]);
}The comments in the source code explain the reasoning behind it. Once a negative start has been adjusted, it might still be negative, and in that case we replace it with 0. However, if stop is still negative after the adjustment, we don't replace it, leveraging the fact that in that case start > stop, so the output will be correct anyway.
CodeCrafters
Lists Stage 5: List elements (negative indexes)
The code we wrote in this section passes Lists - Stage 5 of the CodeCrafters challenge.
Step 10.6 - Refactor the storage
As mentioned in Step 10.4 it's clearly time to review the implementation of Storage. While the code is solid, its overall organisation is not great: approximately 70% of the file src/storage.rs is taken up by the 25 tests, and every new implementation of a command is likely to add more of them.
This leads to the source code being difficult to navigate, so it is worth exploring alternatives. The solution comes from two features of the Rust compiler:
- The implementation of a
struct can be split into several impl {} blocks, even across multiple files. - A file
somename.rs can load the module somename/somemodule.rs directly with mod somemodule;.
This means that we can create the directory src/storage, split the implementation of Storage into src/storage/expiry.rs, src/storage/get.rs, and so on, and keep the code both tidy and well-organised.
The refactoring is basically a matter of moving code around as it is, so we will see an example below and then leave the rest to the reader. The full set of changes can be seen in the relevant commit on GitHub.
Using expiry functions as an example, we add the new module to the file src/storage.rs, then move the code from implementation and tests into src/storage/expiry.rs.
src/storage.rs
use crate::set::{KeyExpiry, SetArgs};
use crate::storage_result::{StorageError, StorageResult};
use std::cmp::min;
use std::collections::HashMap;
use std::ops::Add;
use std::time::{Duration, SystemTime};
mod expiry;
impl Storage {
// Turns on the storage active expiry.
pub fn set_active_expiry(&mut self, value: bool) {
self.active_expiry = value;
}
// Check all keys with an expiry time.
// If the key has expired remove it from the storage.
pub fn expire_keys(&mut self) {
if !self.active_expiry {
return;
}
let now = SystemTime::now();
let expired_keys: Vec<String> = self
.expiry
.iter()
.filter_map(|(key, &value)| if value < now { Some(key.clone()) } else { None })
.collect();
for k in expired_keys {
self.store.remove(&k);
self.expiry.remove(&k);
}
}
mod tests {
#[test]
fn test_expire_keys() {
let mut storage: Storage = Storage::new();
storage
.set(String::from("akey"), String::from("avalue"), SetArgs::new())
.unwrap();
storage.expiry.insert(
String::from("akey"),
SystemTime::now() - Duration::from_secs(5),
);
storage.expire_keys();
assert_eq!(storage.store.len(), 0);
}
#[test]
fn test_expire_keys_deactivated() {
let mut storage: Storage = Storage::new();
storage.set_active_expiry(false);
storage
.set(String::from("akey"), String::from("avalue"), SetArgs::new())
.unwrap();
storage.expiry.insert(
String::from("akey"),
SystemTime::now() - Duration::from_secs(5),
);
storage.expire_keys();
assert_eq!(storage.store.len(), 1);
}The code is moved without changes, but the new file starts with use super::*; to import what is defined in the main file.
src/storage/expiry.rs
use super::*;
impl Storage {
// Turns on the storage active expiry.
pub fn set_active_expiry(&mut self, value: bool) {
...
}
// Check all keys with an expiry time.
// If the key has expired remove it from the storage.
pub fn expire_keys(&mut self) {
...
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
// Test that the function expire_keys removes
// keys that have an expiry time in the past.
fn test_expire_keys() {
...
}
#[test]
// Test that the function expire_keys doesn't remove
// keys that have an expiry time in the past.
// if active expiry is turned off.
fn test_expire_keys_deactivated() {
...
}
}The same set of changes will be applied to the functions Storage::set (src/storage/set.rs), Storage::get (src/storage/get.rs), Storage::rpush (src/storage/rpush.rs), and Storage::lrange (src/storage/lrange.rs). For each one of them, we need to add the corresponding module to src/storage.rs, and the final version is shown below.
src/storage.rs
use crate::set::{KeyExpiry, SetArgs};
use crate::storage_result::{StorageError, StorageResult};
use std::cmp::min;
use std::collections::HashMap;
use std::ops::Add;
use std::time::{Duration, SystemTime};
mod expiry;
mod get;
mod lrange;
mod rpush;
mod set;The only test left in src/storage.rs will be test_create_new.
CodeCrafters
Lists Stage 5: List elements (negative indexes)
The code we wrote in this section is a refactoring of the previous step, so it still passes Lists - Stage 5 of the CodeCrafters challenge.
Step 10.7 - Prepend elements
The implementation of LPUSH is not too different from the implementation of RPUSH in terms of overall structure. Clearly, the core effect of the command is different and requires different code and tests, but most of the code is, if not the same, extremely similar.
We start with the implementation of Storage::lpush as we did in the rest of the chapter. This time, however, we will do it directly in a module inside src/storage. The tests are pretty standard, but we need to add one to check the order of the inserted items, since LPUSH prepends elements one by one, so a call LPUSH key a b c leaves them in the list in the order c, b, a.
src/storage/lpush.rs
#[cfg(test)]
mod tests {
use super::*;
#[test]
// Test that the function lpush works as expected
// when the key doesn't exist.
// The key is created as a list and the value is
// added to it.
fn test_lpush_key_does_not_exist() {
let mut storage: Storage = Storage::new();
let avalue = StorageData::list(["avalue"]);
let output = storage.lpush("akey", &[String::from("avalue")]).unwrap();
assert_eq!(output, 1);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &avalue),
None => panic!(),
}
}
#[test]
// Test that the function lpush works as expected
// when inserting elements.
// Elements are prepended in reverse order, so
// `lpush a b c` prepends `c b a`.
fn test_lpush_order() {
let mut storage: Storage = Storage::new();
let expected = StorageData::list(["c", "b", "a"]);
let output = storage
.lpush(
"akey",
&[String::from("a"), String::from("b"), String::from("c")],
)
.unwrap();
assert_eq!(output, 3);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &expected),
None => panic!(),
}
}
#[test]
// Test that the function lpush works as expected
// when the key exists and is a list.
// The value is inserted at the start.
fn test_lpush_key_exists_and_is_list() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::list(["value1"]));
let list = StorageData::list(["value0", "value1"]);
let output = storage.lpush("akey", &[String::from("value0")]).unwrap();
assert_eq!(output, 2);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &list),
None => panic!(),
}
}
#[test]
// Test that the function lpush
// returns the correct error when
// the key exists and is not a list.
fn test_lpush_key_exists_and_is_not_list() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::string("avalue"));
let error = storage.lpush("akey", &[String::from("value2")]);
assert_eq!(error, Err(StorageError::WrongType));
}
}Once again, the structure of the function is similar to that of Storage::rpush and Storage::lrange, with differences in the code of the branches.
src/storage/lpush.rs
use super::*;
impl Storage {
// Implement the `lpush` operation for the storage.
pub fn lpush(&mut self, key: &str, values: &[String]) -> StorageResult<usize> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let length = match self.store.get_mut(key) {
// The value is a string, we cannot
// use lpush on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => return Err(StorageError::WrongType),
// The value is already a list. Add
// the new element and return the
// length of the whole list.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
// Create an iterator of &String, reverse it,
// own each element cloning into String.
// Add elements at index 0.
v.splice(0..0, values.iter().rev().cloned());
v.len()
}
// The key doesn't exist. Create it,
// add all elements in reversed order,
// and return its length.
None => {
// Create an iterator of &String, reverse it,
// own each element cloning into String.
// Collect the iterator into a vector.
let reversed: Vec<String> = values.iter().rev().cloned().collect();
self.store
.insert(key.to_string(), StorageData::list(reversed));
values.len()
}
};
Ok(length)
}
}The module needs to be linked to the definition of Storage.
src/storage.rs
mod expiry;
mod get;
mod lpush;
mod lrange;
mod rpush;
mod set;
With this code in place, we can export the command in the server. The implementation is pretty standard, but you will notice that StorageError::WrongType is handled explicitly. This is not required by the challenge, but reflects the behaviour of a real Redis implementation, so it has been introduced here and will be ported to the other list commands later in this step.
src/commands/lpush.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
use crate::storage_result::StorageError;
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Extract the storage from the server.
let storage = match server.storage.as_mut() {
Some(storage) => storage,
None => {
request.error(ServerError::StorageNotInitialised).await;
return;
}
};
// Check that the command received at least 2 arguments.
if command.len() < 3 {
request
.error(ServerError::CommandSyntaxError(command.join(" ")))
.await;
return;
}
// Extract the key and the values.
let key = &command[1];
let values = &command[2..];
let result = storage.lpush(key, values);
match result {
Ok(size) => {
request
.data(ServerValue::Resp(Resp::Integer(size as i64)))
.await
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}
}As usual, we can add tests for the requirements of this specific function. We test single and multiple input values, the absence of storage in the server, and the case of a wrong number of parameters.
src/commands/lpush.rs
#[cfg(test)]
mod tests {
use super::*;
use crate::server_result::ServerMessage;
use crate::storage::Storage;
use tokio::sync::mpsc;
#[tokio::test]
// Test that the function command processes
// an `LPUSH` request with a single value.
async fn test_command_single_value() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("lpush"),
String::from("key"),
String::from("value"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::Integer(1)))
);
}
#[tokio::test]
// Test that the function command processes
// an `LPUSH` request with multiple values.
async fn test_command_multiple_values() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("lpush"),
String::from("key"),
String::from("value1"),
String::from("value2"),
String::from("value3"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::Integer(3)))
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called on a server that
// has no storage attached.
async fn test_command_no_storage() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let cmd = vec![
String::from("lpush"),
String::from("key"),
String::from("value"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::StorageNotInitialised)
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called with the wrong
// number of parameters.
async fn test_command_wrong_number_of_parameters() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![String::from("lpush"), String::from("key")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::CommandSyntaxError(String::from("lpush key")))
);
}
}To activate the command we need to expose the module to the compiler.
src/commands/mod.rs
pub mod discard;
pub mod echo;
pub mod exec;
pub mod get;
pub mod incr;
pub mod info;
pub mod lpush;
pub mod lrange;
pub mod multi;
pub mod ping;
pub mod psync;
pub mod replconf;
pub mod rpush;
pub mod set;
pub mod wait;
Last, we have to import it and associate it with a command name.
src/server.rs
use crate::client::Client;
use crate::commands::{
discard, echo, exec, get, incr, info, lrange, multi, ping, psync, replconf, rpush, set, wait,
discard, echo, exec, get, incr, info, lpush, lrange, multi, ping, psync, replconf, rpush, set,
wait,
};
use crate::connection::{
stream_read_data_length, stream_read_line, stream_send_receive_resp, ConnectionMessage,
};
use crate::replication::ReplicationConfig;
pub async fn process_request(request: Request, server: &mut Server) {
"info" => {
info::command(server, &request, &command).await;
}
"lpush" => {
lpush::command(server, &request, &command).await;
}
"lrange" => {
lrange::command(server, &request, &command).await;
}Before we conclude this step, we can extend the explicit handling of StorageError::WrongType to RPUSH and LRANGE.
src/commands/lrange.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
use crate::storage_result::StorageError;
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
let result = storage.lrange(key, start, stop);
match result {
Ok(v) => {
let resp: Vec<Resp> = v.iter().cloned().map(Resp::BulkString).collect();
request.data(ServerValue::Resp(Resp::Array(resp))).await
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
} src/commands/rpush.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
use crate::storage_result::StorageError;
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
let result = storage.rpush(key, value);
match result {
Ok(size) => {
request
.data(ServerValue::Resp(Resp::Integer(size as i64)))
.await
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}CodeCrafters
Lists Stage 6: Prepend elements
The code we wrote in this section passes Lists - Stage 6 of the CodeCrafters challenge.
Step 10.8 - Query list length
The command LLEN is, in terms of requirements, the simplest command on lists. The only special behaviour is that the returned length must be 0 when the list doesn't exist.
Let's implement the method Storage::llen first.
src/storage/llen.rs
use super::*;
impl Storage {
// Implement the `llen` operation for the storage.
pub fn llen(&self, key: &str) -> StorageResult<usize> {
// Get an immutable reference to the value
// of the key and act according to its
// nature.
let length = match self.store.get(key) {
// The value is a string, we cannot
// use llen on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => return Err(StorageError::WrongType),
// The value is a list.
// Return its length.
Some(StorageData {
value: StorageValue::List(v),
..
}) => v.len(),
// The key doesn't exist.
// Return 0
None => 0,
};
Ok(length)
}
}The tests for this method are straightforward.
src/storage/llen.rs
#[cfg(test)]
mod tests {
use super::*;
#[test]
// Test that the function llen works as expected
// when the key doesn't exist.
fn test_llen_key_does_not_exist() {
let storage: Storage = Storage::new();
let output = storage.llen("akey").unwrap();
assert_eq!(output, 0);
assert_eq!(storage.store.len(), 0);
}
#[test]
// Test that the function llen works as expected
// when the key exists and is a list.
fn test_llen_key_exists_and_is_list() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1"]),
);
let output = storage.llen("akey").unwrap();
assert_eq!(output, 2);
assert_eq!(storage.store.len(), 1);
}
#[test]
// Test that the function llen
// returns the correct error when
// the key exists and is not a list.
fn test_llen_key_exists_and_is_not_list() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::string("avalue"));
let error = storage.llen("akey");
assert_eq!(error, Err(StorageError::WrongType));
}
}The module must be added to src/storage.rs.
src/storage.rs
mod expiry;
mod get;
mod llen;
mod lpush;
mod lrange;
mod rpush;
mod set;
Now we can implement the server command.
src/commands/llen.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
use crate::storage_result::StorageError;
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {|@+||@+|
// Extract the storage from the server.
let storage = match server.storage.as_mut() {
Some(storage) => storage,
None => {
request.error(ServerError::StorageNotInitialised).await;
return;
}
};
// Check that the command received 1 argument.
if command.len() != 2 {
request
.error(ServerError::CommandSyntaxError(command.join(" ")))
.await;
return;
}
// Extract the key.
let key = &command[1];
let result = storage.llen(key);
match result {
Ok(size) => {
request
.data(ServerValue::Resp(Resp::Integer(size as i64)))
.await
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}
}The tests for this function are now almost standard.
src/commands/llen.rs
#[cfg(test)]
mod tests {
use super::*;
use crate::server_result::ServerMessage;
use crate::storage::Storage;
use tokio::sync::mpsc;
#[tokio::test]
// Test that the function command processes
// an `LLEN` request.
async fn test_command() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![String::from("llen"), String::from("key")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::Integer(0)))
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called on a server that
// has no storage attached.
async fn test_command_no_storage() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let cmd = vec![String::from("llen"), String::from("key")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::StorageNotInitialised)
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called with the wrong
// number of parameters.
async fn test_command_wrong_number_of_parameters() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![String::from("llen")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::CommandSyntaxError(String::from("llen")))
);
}
}And as usual we need to expose the module.
src/commands/mod.rs
pub mod discard;
pub mod echo;
pub mod exec;
pub mod get;
pub mod incr;
pub mod info;
pub mod llen;
pub mod lpush;
pub mod lrange;
pub mod multi;
pub mod ping;
pub mod psync;
pub mod replconf;
pub mod rpush;
pub mod set;
pub mod wait;
Last, the command is exposed in server::process_request.
src/server.rs
use crate::client::Client;
use crate::commands::{
discard, echo, exec, get, incr, info, lpush, lrange, multi, ping, psync, replconf, rpush, set,
wait,
discard, echo, exec, get, incr, info, llen, lpush, lrange, multi, ping, psync, replconf, rpush,
set, wait,
};
use crate::connection::{
stream_read_data_length, stream_read_line, stream_send_receive_resp, ConnectionMessage,
};
pub async fn process_request(request: Request, server: &mut Server) {
"info" => {
info::command(server, &request, &command).await;
}
"llen" => {
llen::command(server, &request, &command).await;
}
"lpush" => {
lpush::command(server, &request, &command).await;
}CodeCrafters
Lists Stage 7: Query list length
The code we wrote in this section passes Lists - Stage 7 of the CodeCrafters challenge.
Step 10.9 - Remove an element
The addition of LPOP with support for a single element is fairly straightforward. The real Redis command supports popping multiple elements, and we will implement that in the next step.
For the time being, we need to test four use cases for the new function. One is the case in which the key does not exist and three of them are for existing keys: the value is a list, the value is an empty list, and the value is not a list.
The important requirement here is that the server command returns a null reply (a Null bulk string) when the list is empty or if it doesn't exist. This means that we need to handle those use cases returning something that is not an error. The best way to do that is to assume the function will return StorageResult<Option<String>>, using Ok(None) to represent that condition.
With that design choice in mind, we can start alternating tests and code. The first test covers the case in which the key does not exist.
src/storage/lpop.rs
use super::*;
#[cfg(test)]
mod tests {
use super::*;
#[test]
// Test that the function lpop works as expected
// when the key doesn't exist.
fn test_lpop_key_does_not_exist() {
let mut storage: Storage = Storage::new();
let output = storage.lpop("akey").unwrap();
assert_eq!(output, None);
assert_eq!(storage.store.len(), 0);
}
}As usual, the module needs to be added to src/storage.rs to become visible.
src/storage.rs
mod expiry;
mod get;
mod llen;
mod lpop;
mod lpush;
mod lrange;
mod rpush;
mod set;
The test suite will at this point fail with a compilation error as the method Storage::lpop does not exist. To implement it, it seems only natural to retain the structure we used for the other list functions, with the result of a get_mut processed by a match.
src/storage/lpop.rs
use super::*;
impl Storage {
// Implement the `lpop` operation for the storage.
pub fn lpop(&mut self, key: &str) -> StorageResult<Option<String>> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let item = match self.store.get_mut(key) {
// The value is a string, we cannot
// use lpop on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => todo!(),
// The value is a list.
// Return the first element.
Some(StorageData {
value: StorageValue::List(v),
..
}) => todo!(),
// The key doesn't exist.
// Return none.
None => None,
};
Ok(item)
}
}The second test we can write covers the case when the key exists and the value is a list.
src/storage/lpop.rs
mod tests {
#[test]
// Test that the function lpop works as expected
// when the key exists and is a list.
fn test_lpop_key_exists_and_is_list() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1"]),
);
let output = storage.lpop("akey").unwrap();
assert_eq!(output, Some(String::from("value0")));
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &StorageData::list(["value1"])),
None => panic!(),
}
}And the change to the code is straightforward:
src/storage/lpop.rs
impl Storage {
pub fn lpop(&mut self, key: &str) -> StorageResult<Option<String>> {
// The value is a list.
// Return the first element.
Some(StorageData {
value: StorageValue::List(v),
..
}) => todo!(),
}) => {
// Remove and return the first element.
Some(v.remove(0))
}The third test checks the behaviour of the function when the key is an empty list.
src/storage/lpop.rs
mod tests {
#[test]
// Test that the function lpop works as expected
// when the key exists and is an empty list.
fn test_lpop_key_exists_and_is_empty() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::list(["value0"]));
// Pop the only value.
let _ = storage.lpop("akey").unwrap();
let output = storage.lpop("akey").unwrap();
assert_eq!(output, None);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &StorageData::list::<[_; 0], String>([])),
None => panic!(),
}
}The method Vec::remove [docs] panics if the index is out of bounds, so we need to check the state of the vector with v.is_empty().
src/storage/lpop.rs
impl Storage {
pub fn lpop(&mut self, key: &str) -> StorageResult<Option<String>> {
// The value is a list.
// Return the first element.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
// If the list is empty we
// cannot remove elements.
if v.is_empty() {
return Ok(None);
}
// Remove and return the first element.
Some(v.remove(0))
}The last test is for a key that exists but whose value is not a list.
src/storage/lpop.rs
mod tests {
#[test]
// Test that the function lpop
// returns the correct error when
// the key exists and is not a list.
fn test_lpop_key_exists_and_is_not_list() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::string("avalue"));
let error = storage.lpop("akey");
assert_eq!(error, Err(StorageError::WrongType));
}And that covers the last todo!() that we added to the function.
src/storage/lpop.rs
impl Storage {
pub fn lpop(&mut self, key: &str) -> StorageResult<Option<String>> {
// The value is a string, we cannot
// use lpop on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => todo!(),
}) => return Err(StorageError::WrongType),With this in place, we can add the command to the server. Once again, the implementation is straightforward. As mentioned before, the function returns an Option that adds a new branch to the logic of the command.
src/commands/lpop.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
use crate::storage_result::StorageError;
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Extract the storage from the server.
let storage = match server.storage.as_mut() {
Some(storage) => storage,
None => {
request.error(ServerError::StorageNotInitialised).await;
return;
}
};
// Check that the command received 1 argument.
if command.len() != 2 {
request
.error(ServerError::CommandSyntaxError(command.join(" ")))
.await;
return;
}
// Extract the key.
let key = &command[1];
let result = storage.lpop(key);
match result {
Ok(None) => request.data(ServerValue::Resp(Resp::Null)).await,
Ok(Some(elem)) => {
request
.data(ServerValue::Resp(Resp::BulkString(elem)))
.await
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}
}The tests for this function are the following:
src/commands/lpop.rs
#[cfg(test)]
mod tests {
use super::*;
use crate::server_result::ServerMessage;
use crate::storage::Storage;
use tokio::sync::mpsc;
#[tokio::test]
// Test that the function command processes
// an `LPOP` request.
async fn test_command() {
let mut storage = Storage::new();
storage
.rpush("key", &[String::from("value0"), String::from("value1")])
.unwrap();
let mut server: Server = Server::new("localhost".to_string(), 6379);|@+||@+|
server.set_storage(storage);
let cmd = vec![String::from("lpop"), String::from("key")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::bulk_string("value0")))
);
}
#[tokio::test]
// Test that the function command processes
// an `LPOP` request.
async fn test_command_empty_list() {
let mut storage = Storage::new();
storage.rpush("key", &[]).unwrap();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![String::from("lpop"), String::from("key")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::Null))
);
}
#[tokio::test]
// Test that the function command processes
// an `LPOP` request.
async fn test_command_key_does_not_exist() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![String::from("lpop"), String::from("key")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::Null))
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called on a server that
// has no storage attached.
async fn test_command_no_storage() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let cmd = vec![String::from("lpop"), String::from("key")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::StorageNotInitialised)
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called with the wrong
// number of parameters.
async fn test_command_wrong_number_of_parameters() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![String::from("lpop")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::CommandSyntaxError(String::from("lpop")))
);
}
}As we did for all other commands, we need to expose the function and add it to server::process_request.
src/commands/mod.rs
pub mod discard;
pub mod echo;
pub mod exec;
pub mod get;
pub mod incr;
pub mod info;
pub mod llen;
pub mod lpop;
pub mod lpush;
pub mod lrange;
pub mod multi;
pub mod ping;
pub mod psync;
pub mod replconf;
pub mod rpush;
pub mod set;
pub mod wait;
src/server.rs
use crate::client::Client;
use crate::commands::{
discard, echo, exec, get, incr, info, llen, lpush, lrange, multi, ping, psync, replconf, rpush,
set, wait,
discard, echo, exec, get, incr, info, llen, lpop, lpush, lrange, multi, ping, psync, replconf,
rpush, set, wait,
};
use crate::connection::{
stream_read_data_length, stream_read_line, stream_send_receive_resp, ConnectionMessage,
};
use crate::replication::ReplicationConfig;
pub async fn process_request(request: Request, server: &mut Server) {
"llen" => {
llen::command(server, &request, &command).await;
}
"lpop" => {
lpop::command(server, &request, &command).await;
}
"lpush" => {
lpush::command(server, &request, &command).await;
}CodeCrafters
Lists Stage 8: Remove an element
The code we wrote in this section passes Lists - Stage 8 of the CodeCrafters challenge.
Step 10.10 - Remove multiple elements
As discussed in the previous step, the new requirements for LPOP change the function in a relevant way. The command must support an optional value that specifies the number of elements to extract from the list, and the type of the result changes according to the number of items: for a single item, it returns a bulk string (as in the previous step), but for multiple items it returns an array.
To implement this we need to change both the interface of Storage::lpop and its internal logic. The interface change is going to affect the code we already wrote, so we will first migrate the existing logic and then, when Storage::lpop and commands::lpop::command are once again in sync, we will add the new logic to return multiple elements.
Step 10.10.1 - Change interface
The command LPOP returns a single element as a bulk string and multiple elements as an array. There are two ways to implement this: we can always return a list of values from storage::lpop and to change the output of commands::lpop::command according to the length of the list, or return a custom enum from storage::lpop with three different variants: empty, single, multiple.
The first option looks better, particularly because it allows us to implement the very generic approach "extract N elements" that is valid for N >= 0. This simplifies the logic of the internal function and delegates the presentation of the results to the external one. This also looks like a good separation of concerns between the two layers of the system.
src/storage/lpop.rs
use super::*;
impl Storage {
// Implement the `lpop` operation for the storage.
pub fn lpop(&mut self, key: &str) -> StorageResult<Option<String>> {
pub fn lpop(&mut self, key: &str) -> StorageResult<Vec<String>> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let item = match self.store.get_mut(key) {
// The value is a string, we cannot
// use lpop on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => return Err(StorageError::WrongType),
// The value is a list.
// Return the first element.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
// If the list is empty we
// cannot remove elements.
// Return an empty vector.
if v.is_empty() {
return Ok(None);
return Ok(vec![]);
}
// Remove and return the first element.
Some(v.remove(0))
vec![v.remove(0)]
}
// The key doesn't exist.
// Return none.
None => None,
// Return an empty vector.
None => vec![],
};
Ok(item)
}
}The changes to the interface have an impact on tests.
src/storage/lpop.rs
mod tests {
#[test]
// Test that the function lpop works as expected
// when the key doesn't exist.
fn test_lpop_key_does_not_exist() {
let mut storage: Storage = Storage::new();
let output = storage.lpop("akey").unwrap();
assert_eq!(output, None);
assert_eq!(output, Vec::<String>::new());
assert_eq!(storage.store.len(), 0);
}
#[test]
// Test that the function lpop works as expected
// when the key exists and is a list.
fn test_lpop_key_exists_and_is_list() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1"]),
);
let output = storage.lpop("akey").unwrap();
assert_eq!(output, Some(String::from("value0")));
assert_eq!(output, vec![String::from("value0")]);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &StorageData::list(["value1"])),
None => panic!(),
}
}
#[test]
// Test that the function lpop works as expected
// when the key exists and is an empty list.
fn test_lpop_key_exists_and_is_empty() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::list(["value0"]));
// Pop the only value.
let _ = storage.lpop("akey").unwrap();
let output = storage.lpop("akey").unwrap();
assert_eq!(output, None);
assert_eq!(output, Vec::<String>::new());
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &StorageData::list::<[_; 0], String>([])),
None => panic!(),
}
}Type inference
It is interesting to note that the compiler can infer the proper type of the vector items in the function but not in the tests, and the difference is worth a short explanation.
In the function's body, the compiler knows the intended result type StorageResult<Vec<String>> and so it can infer that the macro in Ok(vec![]) actually creates a Vec<String>.
In the tests, instead, the code assert_eq!(output, vec![]); wouldn't work. The reason is that the two values output: Vec<String> and the unspecified Vec<T> are compared using the trait PartialEq, which is implemented by String for many other types. So, T here might be String as well as something else like Bytes, and the compiler cannot decide it autonomously, which is why we need to be more explicit and use the full form Vec::<String>::new().
The change of the function signature must be matched in the server command. We can see the separation of concerns in action here: while Storage::lpop simply outputs a vector, the server command must check if that is empty or not and send to the client two different types of result.
src/commands/lpop.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
let result = storage.lpop(key);
match result {
Ok(None) => request.data(ServerValue::Resp(Resp::Null)).await,
Ok(Some(elem)) => {
request
.data(ServerValue::Resp(Resp::BulkString(elem)))
.await
}
Ok(v) => {
if v.is_empty() {
request.data(ServerValue::Resp(Resp::Null)).await;
return;
}
if v.len() == 1 {
request
.data(ServerValue::Resp(Resp::BulkString(v[0].clone())))
.await
}
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}Step 10.10.2 - Implement new requirements
Now that the code works with the new interface we can implement the new behaviour that returns multiple elements. Let's first have a look at Storage::lpop, where we need to add a new parameter for the number of elements that we want to pop from the list.
src/storage/lpop.rs
use super::*;
impl Storage {
// Implement the `lpop` operation for the storage.
pub fn lpop(&mut self, key: &str) -> StorageResult<Vec<String>> {
pub fn lpop(&mut self, key: &str, num_items: i64) -> StorageResult<Vec<String>> {
// Get a mutable reference to the value
// of the key and act according to its
// nature.
let item = match self.store.get_mut(key) {
// The value is a string, we cannot
// use lpop on it.
Some(StorageData {
value: StorageValue::String(_),
..
}) => return Err(StorageError::WrongType),
// The value is an empty list.
// Return an empty vector.
Some(StorageData {
value: StorageValue::List(v),
..
}) if v.is_empty() => vec![],
// The value is a list.
// Return the first element.
// Return the requested elements.
Some(StorageData {
value: StorageValue::List(v),
..
}) => {
// If the list is empty we
// cannot remove elements.
// Return an empty vector.
if v.is_empty() {
return Ok(vec![]);
}
// Remove and return the first element.
vec![v.remove(0)]
// Make sure the provided number of
// items can be actually fetched.
let max = (num_items.max(0) as usize).min(v.len());
// Extract the items.
v.drain(0..max).collect()
}
// The key doesn't exist.
// Return an empty vector.
None => vec![],
};
Ok(item)
}
}As you can see, the case where the key contains a list has been split into two different cases using the guard if v.is_empty(). This is just a personal choice that keeps the code inside every branch small and readable.
The addition of a parameter and the changes to the logic require adjustments to the existing tests.
src/storage/lpop.rs
mod tests {
#[test]
// Test that the function lpop works as expected
// when the key doesn't exist.
fn test_lpop_key_does_not_exist() {
let mut storage: Storage = Storage::new();
let output = storage.lpop("akey").unwrap();
let output = storage.lpop("akey", 1).unwrap();
assert_eq!(output, Vec::<String>::new());
assert_eq!(storage.store.len(), 0);
}
#[test]
// Test that the function lpop works as expected
// when the key exists and is an empty list.
fn test_lpop_key_exists_and_is_empty() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::list(["value0"]));
// Pop the only value.
let _ = storage.lpop("akey").unwrap();
let output = storage.lpop("akey").unwrap();
let _ = storage.lpop("akey", 1).unwrap();
let output = storage.lpop("akey", 1).unwrap();
assert_eq!(output, Vec::<String>::new());
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &StorageData::list::<[_; 0], String>([])),
None => panic!(),
}
}
#[test]
// Test that the function lpop
// returns the correct error when
// the key exists and is not a list.
fn test_lpop_key_exists_and_is_not_list() {
let mut storage: Storage = Storage::new();
storage
.store
.insert(String::from("akey"), StorageData::string("avalue"));
let error = storage.lpop("akey");
let error = storage.lpop("akey", 1);
assert_eq!(error, Err(StorageError::WrongType));
}We can also rename the test_lpop_key_exists_and_is_list to highlight the fact that it is testing specifically the extraction of a single item. We can then add two new tests to check the behaviour of the function when we extract respectively zero and multiple items.
src/storage/lpop.rs
mod tests {
#[test]
// Test that the function lpop works as expected
// when the key exists and is a list.
// We pop a single item.
fn test_lpop_key_exists_and_is_list() {
fn test_lpop_key_exists_and_is_list_pop_single_item() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1"]),
);
let output = storage.lpop("akey").unwrap();
let output = storage.lpop("akey", 1).unwrap();
assert_eq!(output, vec![String::from("value0")]);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &StorageData::list(["value1"])),
None => panic!(),
}
}
#[test]
// Test that the function lpop works as expected
// when the key exists and is a list.
// We pop zero items.
fn test_lpop_key_exists_and_is_list_pop_zero_items() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1"]),
);
let output = storage.lpop("akey", 0).unwrap();
assert_eq!(output, Vec::<String>::new());
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &StorageData::list(["value0", "value1"])),
None => panic!(),
}
}
#[test]
// Test that the function lpop works as expected
// when the key exists and is a list.
// We pop multiple items.
fn test_lpop_key_exists_and_is_list_pop_multiple_items() {
let mut storage: Storage = Storage::new();
storage.store.insert(
String::from("akey"),
StorageData::list(["value0", "value1", "value2"]),
);
let output = storage.lpop("akey", 2).unwrap();
assert_eq!(output, vec![String::from("value0"), String::from("value1")]);
assert_eq!(storage.store.len(), 1);
match storage.store.get("akey") {
Some(value) => assert_eq!(value, &StorageData::list(["value2"])),
None => panic!(),
}
}The change to the signature of Storage::lpop affects the server command as well. The biggest change, however, comes from the fact that we need to provide a default value for the amount of items fetched by LPOP. As command parameters are strings and Storage::lpop wants an i64 we also need to handle the conversion and the potential errors it can return.
src/commands/lpop.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Extract the storage from the server.
let storage = match server.storage.as_mut() {
Some(storage) => storage,
None => {
request.error(ServerError::StorageNotInitialised).await;
return;
}
};
// Check that the command received 1 argument.
if command.len() != 2 {
// Check that the command received either 1 or 2 arguments.
if command.len() < 2 || command.len() > 3 {
request
.error(ServerError::CommandSyntaxError(command.join(" ")))
.await;
return;
}
// Extract the key.
let key = &command[1];
let result = storage.lpop(key);
// Make sure we have a default
// number of elements to pop.
let num_items_str = &command.get(2).map_or("1", |v| v).to_string();
// Convert the number of elements
// from string to number.
let num_items: i64 = match num_items_str.parse() {
Ok(n) => n,
// The value is not an integer.
Err(_) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"ERR num value is not an integer or out of range",
)))
.await;
return;
}
};
let result = storage.lpop(key, num_items);
match result {
Ok(v) => {
if v.is_empty() {
request.data(ServerValue::Resp(Resp::Null)).await;
return;
}
if v.len() == 1 {
request
.data(ServerValue::Resp(Resp::BulkString(v[0].clone())))
.await
.await;
return;
}
request.data(ServerValue::Resp(Resp::bulk_array(v))).await
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}
}The expression command.get(2) returns an Option<&String>: it is None when the caller did not pass a third argument, or Some(v) when they did. The code map_or("1", |v| v) handles both cases: it returns the default value "1" when the option is None, or applies the closure |v| v (returning v unchanged) when it is Some(v). We could also use the alternative map_or_else, that works in the same way but accepts a closure for the default value instead of a plain value.
The external interface of this function hasn't changed, so the existing tests are still working. We can however add a new test to check how the code handles failures in the aforementioned conversion of the LPOP optional parameter.
src/commands/lpop.rs
mod tests {
#[tokio::test]
// Test that the function command returns the
// correct error when the number of elements
// is not a number.
async fn test_command_number_elements_not_number() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("lpop"),
String::from("key"),
String::from("apple"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::simple_error(
"ERR num value is not an integer or out of range"
)))
);
}CodeCrafters
Lists Stage 9: Remove multiple elements
The code we wrote in this section passes Lists - Stage 9 of the CodeCrafters challenge.
Step 10.11 - Blocking retrieval
So far, the implementation of lists has been straightforward, with no major features added to the system to support the new commands. The next two stages of the challenge present a more complicated scenario.
The command BLPOP is a blocking operation, where the client waits for a list to contain a least one element and then pops it. The command allows the user to specify a timeout for the operation, so the client will be blocked only for a certain amount of time before giving up and returning an empty result if no elements are added to the list in the meantime.
The timeout feature is what adds complexity to the system. It is relatively simple to implement a client that blocks indefinitely: if the list is empty, the client issuing BLPOP is registered in a global queue by the server and as soon as elements are added to the list clients are served on a first-come, first-served basis (FIFO).
However, clients can actually give up, which means they should have a way to leave the queue. This is one of the classic examples where race conditions might occur, so it is worth discussing it in detail.
State and race conditions
Let's imagine we structure the client using a strategy similar to what we implemented for transactions. When the client sends a BLPOP command with a non-zero timeout the server spawns an asynchronous task that does two things: wait for a timeout to expire and wait for an element to appear in the list. If the timeout expires, the client stops waiting for the list to be populated and returns an empty result, but if an element appears in the list the task will pop it and return it to the client.
The first evidence is that tasks cannot monitor the list themselves. As soon as two or more different actors share the same resource there has to be some mechanism for atomic access or for locking to prevent race conditions. If two clients monitor the same list they might both notice that an element is available, and when one of them pops the element the other runs into an error.
However, as we saw in the first part of the book, locking mechanisms can quickly become unmanageable. This leads to the conclusion that the list (as we did for the storage component) should be monitored by the server. The resource is owned by a single entity that will perform one operation at a time, and the server will keep the aforementioned FIFO list of clients to serve.
More race conditions
At this point, we might be tempted to implement the tasks this way: each BLPOP asynchronous task waits for either the timeout to expire or for a message from the server notifying that an element has been added to the list. Upon each event, the task reacts with a specific action, leaving the queue or popping the element.
Once again, however, we have a potential race condition. Let's imagine the server notifies the first waiting task that an element is ready to be popped. The task receives the message, but before it can actually remove it from the list another client issues a standard LPOP, emptying the list and causing a failure of the task.
There are several other "unlucky" combinations of events that can lead to an invalid state of the system. All these should be avoided even though they might be perceived as improbable. As we mentioned above, the fact that the server owns the storage gives us a hint of how to implement this feature keeping the state consistent at all times.
The solution is rather simple: state should be entirely managed by a single owner. In this case, the state we are talking about is the list and its content and the owner has to be the server (or, actually, the storage inside the server).
The server does one thing at a time and therefore cannot run into a race condition by itself. The solution we will implement has to be designed around this idea of the server as the central operator that rules access to the storage, timeouts, and the clients queue.
One of the most classic trade-offs in software and mechanical engineering is that between consistency and parallelism. A system can be made extremely performant splitting its overall job into smaller parallel tasks, but this comes at the cost of an increased risk of inconsistency. The simplest way to be consistent is to reduce parallelism and to make sure that all resources are controlled by the same global entity, but this comes at the cost of lower efficiency.
We see a classic example here: if the server is in charge of controlling the list, increasing the number of clients that access a single list puts a heavy strain on the server, which will have to dedicate an increasing amount of resources to monitoring the queue and the list itself. To be more precise, the trade-off is a three-way one between consistency, throughput. and complexity. Fine-grained locking, for example, can give you both consistency and parallelism, but the cost is an increased complexity of the system.
Moreover, it's important to notice that Redis is I/O bound for most of its operations, which means that most of the time is spent doing things like reading bytes off a socket, parsing RESP, waiting for client responses. State mutations like the ones we are implementing on lists are usually extremely fast in this context. Here, the single-owner paradigm introduces a certain degree of slowness, but in the part that impacts the least on the overall performance of the system.
For those who are interested, an alternative solution is represented by sharding, which is implemented by Redis Cluster. If you want to learn more about sharding and other techniquest, I highly recommend to read the book Designing Data-Intensive Applications by Martin Kleppmann.
The proposed architecture
The system that we are going to implement works according to the following rules:
- A client that issues a
BLPOP on an empty list spawns an asynchronous task that monitors a timeout. The (list, client) tuple is added to a FIFO queue owned by the server. - The task waits for a timeout to expire. When that happens the task notifies the server and stops.
- Whenever a value is added to a list, the server will check the list of blocked clients. If a client is associated with the list, the server will pop the element and send it as a response.
- Whenever the server receives a timeout from a task, it will simply remove the associated
(list, client) from the queue.
As you might realise, this can still lead to small time drifts, but not to race conditions. For example, a task might expire and notify the server, but in the meantime the server might detect an element in the list and send it to the client. At that point, the server can ignore the timeout message as the client has already been removed from the list.
There are a couple of details to clarify before we jump into the implementation:
- If a client issues
BLPOP with timeout 0 nothing of what we described above happens. The command handler will check the list and is it is empty it will immediately return the appropriate response (a RESP null array). - Tasks spawned by the server can communicate with the server itself using the
self_sender mechanism that we implemented for EXEC. The sever main loop picks up the message and processes it in a safe sequential way.
Following what we just described we will tackle the challenge in three phases:
- Implement the necessary infrastructure changes. Create a structure to manage blocked clients, make sure the server uses it when another client sends
RPUSH/LPUSH. - Implement
BLPOP with indefinite blocking and pass stage 10 of the CodeCrafters challenge. - Implement the new version of
BLPOP with full timeout support.
Step 10.11.1 - A structure for blocked clients
Let's add a simple structure into Server that tracks blocked clients. Clients have an id and for convenience we can store a clone of the client sender. The structure is nested: outside we have a HashMap that connects the storage key (the name of the list) to a group of blocked clients. The latter is a VecDeque, a double-ended queue that is the standard queue in Rust. In this case, even though we do not need to work at both ends of the queue, we take advantage of the internal structure that gives us O(1) performance when we take out an element instead of the O(n) of a standard Vec.
Data structures and performance
Say something
src/server.rs
use crate::Resp;
use std::collections::HashMap;
use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
struct BlockedClient {
client_id: u64,
sender: mpsc::Sender<ServerMessage>,
}
pub struct Server {
pub info: ServerInfo,
pub storage: Option<Storage>,
pub replication: ReplicationConfig,
pub replica_senders: Vec<mpsc::Sender<ServerMessage>>,
pub wait_handler_sender: Option<mpsc::Sender<ServerMessage>>,
pub self_sender: Option<mpsc::Sender<ConnectionMessage>>,
pub next_client_id: AtomicU64,
pub clients: HashMap<u64, Client>,
blocked_clients: HashMap<String, VecDeque<BlockedClient>>,
}
impl Server {
pub fn new(host: String, port: u16) -> Self {
Self {
info: ServerInfo { host, port },
storage: None,
replication: ReplicationConfig::new_master(),
replica_senders: Vec::new(),
wait_handler_sender: None,
self_sender: None,
next_client_id: AtomicU64::new(1),
clients: HashMap::new(),
blocked_clients: HashMap::new(),
}
}
pub fn block_client(&mut self, key: &str, client_id: u64, sender: mpsc::Sender<ServerMessage>) {
self.blocked_clients
.entry(key.to_string())
.or_default()
.push_back(BlockedClient { client_id, sender });
}The tests for the function block_client are straightforward.
src/server.rs
mod tests {
#[test]
// Test that the method block_client adds the
// correct data to the internal structure
// of the server.
fn test_block_client_add() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let (connection_sender, mut connection_receiver) = mpsc::channel::<ServerMessage>(32);
server.block_client("key", 1, connection_sender);
let blocked_clients_queue = server.blocked_clients.get("key").unwrap();
let blocked_client = blocked_clients_queue.front().unwrap();
assert_eq!(blocked_client.client_id, 1);
blocked_client
.sender
.try_send(ServerMessage::Data(ServerValue::None))
.unwrap();
assert!(connection_receiver.try_recv().is_ok());
}
#[test]
// Test that the method block_client adds clients
// on the same list using a FIFO strategy.
fn test_block_client_fifo() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let (connection_sender1, mut connection_receiver1) = mpsc::channel::<ServerMessage>(32);
let (connection_sender2, _) = mpsc::channel::<ServerMessage>(32);
server.block_client("key", 1, connection_sender1);
server.block_client("key", 2, connection_sender2);
let blocked_clients_queue = server.blocked_clients.get("key").unwrap();
assert_eq!(blocked_clients_queue.len(), 2);
let blocked_client = blocked_clients_queue.front().unwrap();
assert_eq!(blocked_client.client_id, 1);
blocked_client
.sender
.try_send(ServerMessage::Data(ServerValue::None))
.unwrap();
assert!(connection_receiver1.try_recv().is_ok());
}Step 10.11.2 - A method to notify blocked clients
Now we need to implement a function that notifies waiting clients when a new element is added to a given list. The list is identified by its storage key, so the first thing we need to do is to extract the queue of clients associated with that key and find the first client that is waiting for updates. In case a client is present, we need to proceed with the notification action.
src/server.rs
impl Server {
pub async fn notify_blocked_clients(&mut self, key: &str) {
// Extract the queue associated with the given key.
// The default is a convenient way to deal
// with the case of the key not being registered.
let queue = self.blocked_clients.entry(key.to_string()).or_default();
// Extract the first client that entered the queue.
let first_client = queue.pop_front();
// If the client is present we need to
// pop the first element of the associated
// list and send it to the client.
if let Some(client) = first_client {
// CODE HERE
}
}Inside that block, we need to do several things. First, we need to check that the storage is initialised.
src/server.rs
impl Server {
pub async fn notify_blocked_clients(&mut self, key: &str) {
// If the client is present we need to
// pop the first element of the associated
// list and send it to the client.
if let Some(client) = first_client {
// Make sure the server has an initialised
// storage. If not, we send an error to
// the client.
let Some(storage) = self.storage.as_mut() else {
let _ = client
.sender
.send(ServerMessage::Error(ServerError::StorageNotInitialised))
.await;
return;
};
}Once that is done, we can pop the element from the list. Remember that we are implementing a single-owner scenario, so the server can safely perform this operation without running into issues with other clients adding or removing elements concurrently. After having popped the element, the server needs to send the correct message to the client.
src/server.rs
impl Server {
pub async fn notify_blocked_clients(&mut self, key: &str) {
// If the client is present we need to
// pop the first element of the associated
// list and send it to the client.
if let Some(client) = first_client {
// Make sure the server has an initialised
// storage. If not, we send an error to
// the client.
let Some(storage) = self.storage.as_mut() else {
let _ = client
.sender
.send(ServerMessage::Error(ServerError::StorageNotInitialised))
.await;
return;
};
// Pop the element from the list.
let value = storage.lpop(key, 1);
let message = match value {
// CODE HERE
};
let _ = client.sender.send(message).await;
}There are three main use cases for the extracted value.
- The list is empty. In this case the element has been removed by another client between the detection and the notification. We add the client back to the queue.
- The element has been correctly retrieved. We send the element to the client.
- The key is present but is not a list. We send a specific error to the client.
- The retrieval of the value generates another unspecified error. We send a generic error to the client.
src/server.rs
use crate::server_result::{ServerError, ServerMessage, ServerResult, ServerValue};
use crate::storage::Storage;
use crate::storage_result::StorageError;
use crate::Resp;
use std::collections::{HashMap, VecDeque};
impl Server {
pub async fn notify_blocked_clients(&mut self, key: &str) {
let message = match value {
// The list is empty. A concurrent LPOP might
// have removed the element. Add the client back
// to the queue.
Ok(v) if v.is_empty() => {
queue.push_front(client);
return;
}
// The list returned an element. Send it to
// the client.
Ok(v) => ServerMessage::Data(ServerValue::Resp(Resp::bulk_array([key, &v[0]]))),
// The key is not a list.
Err(StorageError::WrongType) => {
ServerMessage::Data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
}
// Another error occurred while retrieving the key.
Err(_) => ServerMessage::Error(ServerError::CommandInternalError(String::from(
"Error while retrieving the value of the key",
))),
};
let _ = client.sender.send(message).await;
}All the logic we implemented must be verified through tests.
src/server.rs
mod tests {
use super::*;
use crate::server_result::{ServerMessage, ServerValue};
use crate::set::SetArgs;
use crate::storage_result::StorageResult;
#[tokio::test]
// Test that the method notify_blocked_clients
// behaves correctly when no clients are blocked
// on the given key.
async fn test_notify_blocked_clients_no_clients() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let (connection_sender, mut connection_receiver) = mpsc::channel::<ServerMessage>(32);
server.block_client("key1", 1, connection_sender);
server.notify_blocked_clients("key2").await;
assert!(connection_receiver.try_recv().is_err());
}
#[tokio::test]
// Test that the method notify_blocked_clients
// behaves correctly when a single client is blocked
// on the given key.
// The method should run lpop on the given list
// and notify the client.
async fn test_notify_blocked_clients_one_client() {
let mut storage = Storage::new();
let _ = storage.lpush("key1", &[String::from("apple")]);
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let (connection_sender, mut connection_receiver) = mpsc::channel::<ServerMessage>(32);
server.block_client("key1", 1, connection_sender);
server.notify_blocked_clients("key1").await;
assert_eq!(
server.storage.unwrap().lpop("key1", 1),
StorageResult::Ok(vec![])
);
assert_eq!(
connection_receiver.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::bulk_array(["key1", "apple"])))
);
}
#[tokio::test]
// Test that the method notify_blocked_clients
// behaves correctly when two clients are blocked
// on the given key.
// The method should run lpop on the given list
// and notify the first client that entered the queue.
async fn test_notify_blocked_clients_multiple_clients() {
let mut storage = Storage::new();
let _ = storage.lpush("key1", &[String::from("apple")]);
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let (connection_sender1, mut connection_receiver1) = mpsc::channel::<ServerMessage>(32);
let (connection_sender2, mut connection_receiver2) = mpsc::channel::<ServerMessage>(32);
server.block_client("key1", 1, connection_sender1);
server.block_client("key1", 2, connection_sender2);
server.notify_blocked_clients("key1").await;
assert_eq!(
server.storage.unwrap().lpop("key1", 1),
StorageResult::Ok(vec![])
);
assert_eq!(
connection_receiver1.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::bulk_array(["key1", "apple"])))
);
assert!(connection_receiver2.try_recv().is_err());
}
#[tokio::test]
// Test that the method notify_blocked_clients
// sends the correct error message when a client
// must be notified but the storage has not
// been initialised.
async fn test_notify_blocked_clients_no_storage() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let (connection_sender, mut connection_receiver) = mpsc::channel::<ServerMessage>(32);
server.block_client("key1", 1, connection_sender);
server.notify_blocked_clients("key1").await;
assert_eq!(
connection_receiver.try_recv().unwrap(),
ServerMessage::Error(ServerError::StorageNotInitialised)
);
}
#[tokio::test]
// Test that the method notify_blocked_clients
// sends the correct error message when a client
// must be notified but the storage key doesn't
// contain a list.
async fn test_notify_blocked_clients_key_is_not_list() {
let mut storage = Storage::new();
storage
.set("key".to_string(), "value".to_string(), SetArgs::new())
.unwrap();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let (connection_sender, mut connection_receiver) = mpsc::channel::<ServerMessage>(32);
server.block_client("key", 1, connection_sender);
server.notify_blocked_clients("key").await;
assert_eq!(
connection_receiver.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
);
}Step 10.11.3 - Connect RPUSH/LPUSH
Now that we have a function to notify blocked clients, we can call it when an element is added to a list, which means when a client performs an RPUSH or an LPUSH.
src/commands/rpush.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
match result {
Ok(size) => {
request
.data(ServerValue::Resp(Resp::Integer(size as i64)))
.await
.await;
// If any client is waiting on a BLPOP,
// send the first element.
server.notify_blocked_clients(key).await;
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
} src/commands/lpush.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
match result {
Ok(size) => {
request
.data(ServerValue::Resp(Resp::Integer(size as i64)))
.await
.await;
// If any client is waiting on a BLPOP,
// send the first element.
server.notify_blocked_clients(key).await;
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}Step 10.11.4 - Implement BLPOP with no timeout
Finally, we can expose the command BLPOP in its initial version without support for timeout, that will be added in the next step. The timeout present on the command line must be processed to pass the challenge test, but it will be ignored in terms of business logic.
The code of the function command is very similar to other commands that we implemented, and the core logic is the call to server.block_client if the list is empty.
src/commands/blpop.rs
use crate::request::Request;
use crate::resp::Resp;
use crate::server::Server;
use crate::server_result::{ServerError, ServerValue};
use crate::storage_result::StorageError;
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Extract the storage from the server.
let storage = match server.storage.as_mut() {
Some(storage) => storage,
None => {
request.error(ServerError::StorageNotInitialised).await;
return;
}
};
// Check that the command received 3 arguments.
if command.len() != 3 {
request
.error(ServerError::CommandSyntaxError(command.join(" ")))
.await;
return;
}
// Extract the key.
let key = &command[1];
// Extract the timeout.
let timeout_str = &command[2];
// Convert the timeout from string to number.
let _timeout: i64 = match timeout_str.parse() {
Ok(n) => n,
// The value is not an integer.
Err(_) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"ERR timeout is not an integer",
)))
.await;
return;
}
};
// Run a standard lpop on the storage.
let result = storage.lpop(key, 1);
match result {
Ok(v) => {
// If the list is empty we register the
// client in the server, associated with
// the requested key. When the list
// is populated, the server will notify
// the client.
if v.is_empty() {
server.block_client(key, request.client_id, request.sender.clone());
return;
}
request
.data(ServerValue::Resp(Resp::bulk_array([key, &v[0]])))
.await;
}
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.await
}
Err(_) => request.error(ServerError::IncorrectData).await,
}
}At this level of the implementation, we are testing the external behaviour of the server, so we have multiple use cases to cover. The code comments clarify what scenario each test is exploring.
src/commands/blpop.rs
#[cfg(test)]
mod tests {
use super::*;
use crate::server_result::ServerMessage;
use crate::set::SetArgs;
use crate::storage::Storage;
use tokio::sync::mpsc;
#[tokio::test]
// Test that the function command processes
// a `BLPOP` request without timeout.
// The list contains a value.
async fn test_command() {
let mut storage = Storage::new();
storage
.rpush("key", &[String::from("value0"), String::from("value1")])
.unwrap();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("blpop"),
String::from("key"),
String::from("0"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::bulk_array(["key", "value0"])))
);
}
#[tokio::test]
// Test that the function command processes
// a `BLPOP` request.
// The list is empty.
async fn test_command_empty_list() {
let mut storage = Storage::new();
storage.rpush("key", &[]).unwrap();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("blpop"),
String::from("key"),
String::from("0"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
// Wrap the command in a short timeout.
// When the timeout expires, the channel
// should be empty as the command is
// waiting indefinitely.
let _ = tokio::time::timeout(
std::time::Duration::from_millis(50),
command(&mut server, &request, &cmd),
)
.await;
assert!(request_channel_rx.try_recv().is_err());
}
#[tokio::test]
// Test that the function command processes
// a `BLPOP` request.
// The key does not exist, and BLPOP treats
// it as an empty list.
async fn test_command_key_does_not_exist() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("blpop"),
String::from("key"),
String::from("0"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
// Wrap the command in a short timeout.
// When the timeout expires, the channel
// should be empty as the command is
// waiting indefinitely.
let _ = tokio::time::timeout(
std::time::Duration::from_millis(50),
command(&mut server, &request, &cmd),
)
.await;
assert!(request_channel_rx.try_recv().is_err());
}
#[tokio::test]
// Test that the function command returns the
// correct error when called on a server that
// has no storage attached.
async fn test_command_no_storage() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let cmd = vec![
String::from("blpop"),
String::from("key"),
String::from("0"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::StorageNotInitialised)
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called with the wrong
// number of parameters.
// No key is specified.
async fn test_command_wrong_number_of_parameters_no_key() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![String::from("blpop")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::CommandSyntaxError(String::from("blpop")))
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when called with the wrong
// number of parameters.
// No timeout is specified.
async fn test_command_wrong_number_of_parameters_no_timeout() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![String::from("blpop"), String::from("key")];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Error(ServerError::CommandSyntaxError(String::from("blpop key")))
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when the timeout is not a number.
async fn test_command_timeout_not_number() {
let storage = Storage::new();
let mut server: Server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("blpop"),
String::from("key"),
String::from("apple"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::simple_error(
"ERR timeout is not an integer"
)))
);
}
#[tokio::test]
// Test that the function command returns the
// correct error when the value of the key
// is not a list.
async fn test_command_key_is_not_list() {
let mut storage = Storage::new();
storage
.set("key".to_string(), "value".to_string(), SetArgs::new())
.unwrap();
let mut server = Server::new("localhost".to_string(), 6379);
server.set_storage(storage);
let cmd = vec![
String::from("blpop"),
String::from("key"),
String::from("0"),
];
let (request_channel_tx, mut request_channel_rx) = mpsc::channel::<ServerMessage>(32);
let request = Request {
value: Resp::Null,
sender: request_channel_tx.clone(),
binary: Vec::new(),
client_id: 1,
master_connection: false,
};
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value"
)))
);
}
}Before wrapping up the section, the command module must be exposed to the compiler.
src/commands/mod.rs
pub mod blpop;
pub mod discard;
pub mod echo;
pub mod exec;
pub mod get;
pub mod incr;
pub mod info;
pub mod llen;
pub mod lpop;
pub mod lpush;
pub mod lrange;
pub mod multi;
pub mod ping;
pub mod psync;
pub mod replconf;
pub mod rpush;
pub mod set;
pub mod wait;
And last, the command is added to the function process_request to become available to clients.
src/server.rs
use crate::client::Client;
use crate::commands::{
discard, echo, exec, get, incr, info, llen, lpop, lpush, lrange, multi, ping, psync, replconf,
rpush, set, wait,
blpop, discard, echo, exec, get, incr, info, llen, lpop, lpush, lrange, multi, ping, psync,
replconf, rpush, set, wait,
};
use crate::connection::{
stream_read_data_length, stream_read_line, stream_send_receive_resp, ConnectionMessage,
};
pub async fn process_request(request: Request, server: &mut Server) {
"blpop" => {
blpop::command(server, &request, &command).await;
}
"discard" => {
discard::command(server, &request, &command).await;
}
"echo" => {
echo::command(server, &request, &command).await;
}CodeCrafters
Lists Stage 10: Blocking retrieval
The code we wrote in this section passes Lists - Stage 10 of the CodeCrafters challenge.
Step 10.12 - Blocking retrieval with timeout
The final step of the challenge is the implementation of the timeout mechanism in BLPOP. As a reminder, if a client specifies a timeout it means they want to give up after a certain amount of time if the list was not populated. We also decided to separate the management of the timeout from the actual removal of the client: in our architecture (see "The proposed architecture" in 10.11) the timeout monitoring async task notified the server and the server will eventually remove the client. Keep in mind, however, that the two actions are separated and the server might send a value to a client that timed out under certain circumstances.
This feature will be implemented in two separate substeps: removing blocked clients and managing client timeouts.
Step 10.12.1 - Remove blocked clients
In this step, we want to implement a function to remove blocked clients that will be triggered when the timeout expires.
First of all, the challenge test uses a float for the timeout, so the type i64 we used previously must be replaced by an f64 in the command parsing logic. We also adjust comments and error messages to keep everything consistent.
src/commands/blpop.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Convert the timeout from string to number.
let _timeout: i64 = match timeout_str.parse() {
let _timeout: f64 = match timeout_str.parse() {
Ok(n) => n,
// The value is not an integer.
// The value is not a valid number.
Err(_) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"ERR timeout is not an integer",
"ERR timeout is not a number",
)))
.await;
return;
}
};
mod tests {
async fn test_command_timeout_not_number() {
command(&mut server, &request, &cmd).await;
assert_eq!(
request_channel_rx.try_recv().unwrap(),
ServerMessage::Data(ServerValue::Resp(Resp::simple_error(
"ERR timeout is not an integer"
"ERR timeout is not a number"
)))
);
}We want the async task that monitors the client timeout to be able to communicate the expiry to the server. We add a new variant to ConnectionMessage to convey that message.
src/connection.rs
#[derive(Debug)]
pub enum ConnectionMessage {
NewClient(mpsc::Sender<ServerMessage>),
BlockedClientTimeout { key: String, client_id: u64 },
Request(Request),
}We also need a new variant of Resp. On timeout, BLPOP replies with *-1\r\n which is an empty array, distinct from the null string $-1\r\n.
src/resp.rs
#[derive(Debug, PartialEq, Clone)]
pub enum Resp {
Array(Vec<Self>),
BulkString(String),
Integer(i64),
Null,
NullArray,
RDBPrefix(usize),
SimpleError(String),
SimpleString(String),
}
impl fmt::Display for Resp {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Array(data) => {
write!(f, "*{}\r\n", data.len())?;
for elem in data.iter() {
write!(f, "{}", elem)?;
}
Ok(())
}
Self::BulkString(data) => write!(f, "${}\r\n{}\r\n", data.len(), data),
Self::Integer(data) => write!(f, ":{}\r\n", data),
Self::Null => write!(f, "$-1\r\n"),
Self::NullArray => write!(f, "*-1\r\n"),
Self::RDBPrefix(data) => write!(f, "${}\r\n", data),
Self::SimpleError(data) => write!(f, "-{}\r\n", data),
Self::SimpleString(data) => write!(f, "+{}\r\n", data),
}
}
}Last, the implementation of the method remove_blocked_client that the server will call when the timeout message is received. The function extracts from blocked_clients the queue connected with the given key and removes the client with the given client_id if present. Once removed, the client is sent the null array reply.
src/server.rs
impl Server {
pub async fn remove_blocked_client(&mut self, key: &str, client_id: u64) {
// Extract the queue associated with the given key.
// The default is a convenient way to deal
// with the case of the key not being registered.
let queue = self.blocked_clients.entry(key.to_string()).or_default();
if let Some(index) = queue.iter().position(|value| value.client_id == client_id) {
if let Some(blocked_client) = queue.remove(index) {
let _ = blocked_client
.sender
.send(ServerMessage::Data(ServerValue::Resp(Resp::NullArray)))
.await;
}
}
}The function is called inside run_server when the message BlockedClientTimeout is received.
src/server.rs
pub async fn run_server(mut server: Server, mut crx: mpsc::Receiver<ConnectionMessage>) {
match message {
// A new client has connected. Assign an ID
// and store the client in the server.
ConnectionMessage::NewClient(sender) => {
// Get the current ID and atomically
// increase it.
let client_id = server.next_client_id.fetch_add(1, Ordering::Relaxed);
// Create the client.
let client = Client::new(client_id, sender.clone());
// Add the client to the global register.
server.clients.insert(client_id, client);
// Send the client ID to the Connection Handler.
if sender.send(ServerMessage::ClientInit(client_id)).await.is_err() {
eprintln!("Error initialising client");
return;
};
}
// The blocked client timed out and is
// signalling that it wants to be removed
// from the queue.
ConnectionMessage::BlockedClientTimeout {key, client_id} => {
server.remove_blocked_client(&key, client_id).await;
}
// If the message contains a request, extract it.
ConnectionMessage::Request(request) => {
// Process the request.
process_request(request, &mut server).await;
}
}The logic behind the blocked client removal can be checked with two new tests.
src/server.rs
#[tokio::test]
// Test that the method remove_blocked_client
// correctly removes a client added to a queue.
async fn test_remove_blocked_client() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let (connection_sender1, _) = mpsc::channel::<ServerMessage>(32);
let (connection_sender2, _) = mpsc::channel::<ServerMessage>(32);
let (connection_sender3, _) = mpsc::channel::<ServerMessage>(32);
server.block_client("key1", 1, connection_sender1);
server.block_client("key1", 2, connection_sender2);
server.block_client("key2", 3, connection_sender3);
server.remove_blocked_client("key1", 2).await;
let queue1 = server.blocked_clients.get("key1").unwrap();
let queue2 = server.blocked_clients.get("key2").unwrap();
assert_eq!(queue1.len(), 1);
assert_eq!(queue2.len(), 1);
let blocked_client = queue1.front().unwrap();
assert_eq!(blocked_client.client_id, 1);
}
#[tokio::test]
// Test that the method remove_blocked_client
// works when there is no client matching
// the given key and client_id.
async fn test_remove_blocked_client_no_client() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let (connection_sender1, _) = mpsc::channel::<ServerMessage>(32);
let (connection_sender2, _) = mpsc::channel::<ServerMessage>(32);
server.block_client("key1", 1, connection_sender1);
server.block_client("key1", 2, connection_sender2);
server.remove_blocked_client("key2", 3).await;
let queue1 = server.blocked_clients.get("key1").unwrap();
assert_eq!(queue1.len(), 2);
}Step 10.12.2 - Manage blocked client timeout
At this point, we want to add timeout management code to block_client. If the timeout is greater than zero, the function must add the client to the waiting queue blocked_clients and to spawn a task that monitors the timeout itself. As we discussed, the task has a simple job: when the timeout expires it will send a message ConnectionMessage::BlockedClientTimeout to the server.
The first thing to do, then, is to create such handler.
src/server.rs
pub async fn blocked_client_handler(
timeout: f64,
key: String,
client_id: u64,
server_sender: mpsc::Sender<ConnectionMessage>,
) {
tokio::time::sleep(Duration::from_secs_f64(timeout)).await;
let _ = server_sender
.send(ConnectionMessage::BlockedClientTimeout { key, client_id })
.await;
}As you can see, the task is essential. A call to tokio::time::sleep to wait for the timeout and a message to the server.
Next, the task must be spawned in block_client. The function must accept a timeout and we need to clone the server sender when the client is added to the queue. The reason we didn't have to clone before is that the value was dropped without further use (redundant clone), while now the sender will actively be used.
src/server.rs
impl Server {
pub fn block_client(&mut self, key: &str, client_id: u64, sender: mpsc::Sender<ServerMessage>) {
pub fn block_client(
&mut self,
timeout: f64,
key: &str,
client_id: u64,
sender: mpsc::Sender<ServerMessage>,
) {
self.blocked_clients
.entry(key.to_string())
.or_default()
.push_back(BlockedClient { client_id, sender });
.push_back(BlockedClient {
client_id,
sender: sender.clone(),
});
[...]After this, the logic to manage the timeout is straightforward: extract the self sender from the server and spawn the timeout monitoring task.
src/server.rs
impl Server {
pub fn block_client(
&mut self,
timeout: f64,
key: &str,
client_id: u64,
sender: mpsc::Sender<ServerMessage>,
) {
self.blocked_clients
.entry(key.to_string())
.or_default()
.push_back(BlockedClient {
client_id,
sender: sender.clone(),
});
if timeout > 0.0 {
// To monitor a blocked client the server
// needs to have a self sender. This should be
// initialised but as it's optional we need to check.
let server_sender = match &self.self_sender {
Some(sender) => sender.clone(),
None => {
let _ =
sender.try_send(ServerMessage::Error(ServerError::CommandInternalError(
String::from("Server self sender not initialised"),
)));
return;
}
};
// Spawn the blocked client handler.
tokio::spawn(blocked_client_handler(
timeout,
key.to_string(),
client_id,
server_sender,
));
};
}This change affects all tests that call block_client. Such tests must be passed a value of 0.0 as they check the behaviour when there is no timeout, so all of them require the same change. Use cargo build to check which tests must be fixed.
src/server.rs
mod tests {
#[test]
// Test that the method block_client adds the
// correct data to the internal structure
// of the server.
fn test_block_client_add() {
let mut server: Server = Server::new("localhost".to_string(), 6379);
let (connection_sender, mut connection_receiver) = mpsc::channel::<ServerMessage>(32);
server.block_client("key", 1, connection_sender);
server.block_client(0.0, "key", 1, connection_sender);
let blocked_clients_queue = server.blocked_clients.get("key").unwrap();
let blocked_client = blocked_clients_queue.front().unwrap();
assert_eq!(blocked_client.client_id, 1);
blocked_client
.sender
.try_send(ServerMessage::Data(ServerValue::None))
.unwrap();
assert!(connection_receiver.try_recv().is_ok());
}The change to block_client propagates to the command BLPOP where we need to pass the timeout value.
src/commands/blpop.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
// Extract the timeout.
let timeout_str = &command[2];
// Convert the timeout from string to number.
let _timeout: f64 = match timeout_str.parse() {
let timeout: f64 = match timeout_str.parse() {
Ok(n) => n,
// The value is not a valid number.
Err(_) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"ERR timeout is not a number",
)))
.await;
return;
}
};
match result {
// If the list is empty we register the
// client in the server, associated with
// the requested key. When the list
// is populated, the server will notify
// the client.
if v.is_empty() {
server.block_client(key, request.client_id, request.sender.clone());
server.block_client(timeout, key, request.client_id, request.sender.clone());
return;
}CodeCrafters
Lists Stage 11: Blocking retrieval with timeout
The code we wrote in this section passes Lists - Stage 11 of the CodeCrafters challenge.
Step 10.13 - Refactor WRONGTYPE
Since its introduction in step 10.7, the message WRONGTYPE Operation against a key holding the wrong kind of value has become ubiquitous. It is worth refactoring it into a method of Resp as we did for other RESP types to reduce code duplication.
src/resp.rs
impl Resp {
pub fn bulk_string(s: impl Into<String>) -> Self {
Self::BulkString(s.into())
}
pub fn simple_error(s: impl Into<String>) -> Self {
Self::SimpleError(s.into())
}
pub fn simple_string(s: impl Into<String>) -> Self {
Self::SimpleString(s.into())
}
pub fn bulk_array<I, T>(items: I) -> Self
where
I: IntoIterator<Item = T>,
T: Into<String>,
{
Self::Array(items.into_iter().map(Self::bulk_string).collect())
}
pub fn wrong_type_error() -> Self {
Self::SimpleError(String::from(
"WRONGTYPE Operation against a key holding the wrong kind of value",
))
}
}There are 9 different places where we can use the new function, two in src/server.rs and the rest in the list-related commands RPUSH, BLPOP, LPUSH, LRANGE, LLEN, and LPOP. The change is straightforward and is exemplified below.
src/commands/blpop.rs
pub async fn command(server: &mut Server, request: &Request, command: &[String]) {
match result {
Err(StorageError::WrongType) => {
request
.data(ServerValue::Resp(Resp::simple_error(
"WRONGTYPE Operation against a key holding the wrong kind of value",
)))
.data(ServerValue::Resp(Resp::wrong_type_error()))
.await
}CodeCrafters
Lists Stage 11: Blocking retrieval with timeout
Since this last step was a mere refactoring, our code should still pass Lists - Stage 11 of the CodeCrafters challenge.