Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add RDB dump #81

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Add RDB dump #81

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
dd35ecb
Add RDB dump
Morgan279 Sep 27, 2022
65b4485
Clean up redundant code
Morgan279 Sep 28, 2022
b1ce5c3
Use stream scan for wide range scan
yongman Sep 7, 2022
3c1a047
Change to stream scan and fix some defects
Morgan279 Sep 28, 2022
39a5c06
Add configurable rdb file name
Morgan279 Sep 28, 2022
9923d92
Add some TODOs
Morgan279 Sep 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 16 additions & 5 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ slog-term = { version = "2.4" }
tokio = { version = "1", features = ["full"] }
tokio-stream = "0.1"
tokio-util = { version = "0.7.1", features = ["rt"] }
tikv-client = { git = "https://github.com/yongman/client-rust.git", branch = "dev" }
tikv-client = { git = "https://github.com/yongman/client-rust.git", branch = "dev-scanner" }
#tikv-client = { path = "../client-rust" }
lazy_static = "1.4.0"
thiserror = "1"
Expand Down
16 changes: 13 additions & 3 deletions src/cmd/fake.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,14 @@ use tokio::sync::Mutex;
use crate::client::Client;
use crate::cmd::Invalid;
use crate::tikv::errors::{
REDIS_INVALID_CLIENT_ID_ERR, REDIS_NOT_SUPPORTED_ERR, REDIS_NO_SUCH_CLIENT_ERR,
REDIS_VALUE_IS_NOT_INTEGER_ERR,
REDIS_DUMPING_ERR, REDIS_INVALID_CLIENT_ID_ERR, REDIS_NOT_SUPPORTED_ERR,
REDIS_NO_SUCH_CLIENT_ERR, REDIS_VALUE_IS_NOT_INTEGER_ERR,
};
use crate::{
config::LOGGER,
tikv::errors::REDIS_UNKNOWN_SUBCOMMAND,
utils::{resp_bulk, resp_err, resp_int, resp_invalid_arguments, resp_nil, resp_ok},
Connection, Frame, Parse,
Connection, Frame, Parse, RDB,
};

#[derive(Debug, Clone)]
Expand Down Expand Up @@ -232,6 +232,16 @@ impl Fake {
_ => resp_err(REDIS_UNKNOWN_SUBCOMMAND),
}
}
"SAVE" => match RDB::dump().await {
// TODO make it possible to disable this command from configuration file
Ok(()) => resp_ok(),
Err(e) => {
if e != REDIS_DUMPING_ERR {
RDB::reset_dumping();
}
resp_err(e)
}
},
// can not reached here
_ => resp_nil(),
};
Expand Down
7 changes: 7 additions & 0 deletions src/cmd/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,6 +325,7 @@ pub enum Command {
ReadOnly(Fake),
Client(Fake),
Info(Fake),
Save(Fake),

// multi/exec/abort
Multi(Multi),
Expand Down Expand Up @@ -572,6 +573,10 @@ impl Command {
Fake::parse_frames(&mut parse, "info"),
&mut parse,
)),
"save" => Command::Save(transform_parse(
Fake::parse_frames(&mut parse, "save"),
&mut parse,
)),
"multi" => Command::Multi(Multi::new()),
"exec" => Command::Exec(Multi::new()),
"discard" => Command::Discard(Multi::new()),
Expand Down Expand Up @@ -779,6 +784,7 @@ impl Command {
ReadOnly(cmd) => cmd.apply("readonly", dst, cur_client, clients).await,
Client(cmd) => cmd.apply("client", dst, cur_client, clients).await,
Info(cmd) => cmd.apply("info", dst, cur_client, clients).await,
Save(cmd) => cmd.apply("save", dst, cur_client, clients).await,

Scan(cmd) => cmd.apply(dst).await,
Xscan(cmd) => cmd.apply(dst).await,
Expand Down Expand Up @@ -877,6 +883,7 @@ impl Command {
Command::ReadOnly(_) => "readonly",
Command::Client(_) => "client",
Command::Info(_) => "info",
Command::Save(_) => "save",
Command::Multi(_) => "multi",
Command::Exec(_) => "exec",
Command::Discard(_) => "discard",
Expand Down
13 changes: 13 additions & 0 deletions src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ struct Server {
cluster_topology_interval: Option<u64>,
cluster_topology_expire: Option<u64>,
meta_key_number: Option<u16>,
rdb_file_name: Option<String>,
}

#[derive(Debug, Deserialize, Clone)]
Expand Down Expand Up @@ -347,6 +348,18 @@ pub fn config_meta_key_number_or_default() -> u16 {
100
}

pub fn config_rdb_file_name_or_default() -> String {
unsafe {
if let Some(c) = &SERVER_CONFIG {
if let Some(f) = c.server.rdb_file_name.clone() {
return f;
}
}
}

"dump.rdb".to_owned()
}

fn log_level_str() -> String {
unsafe {
if let Some(c) = &SERVER_CONFIG {
Expand Down
28 changes: 14 additions & 14 deletions src/gc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use futures::FutureExt;
use futures::{FutureExt, StreamExt};
use slog::{debug, error, info};
use std::collections::HashSet;
use std::convert::TryInto;
Expand Down Expand Up @@ -107,7 +107,7 @@ impl GcMaster {
let bound_range = KEY_ENCODER.encode_txnkv_gc_version_key_range();

// TODO scan speed throttling
let iter_res = txn.scan(bound_range, u32::MAX).await;
let iter_res = txn.scan_stream(bound_range, u32::MAX).await;
if iter_res.is_err() {
error!(
LOGGER,
Expand All @@ -119,8 +119,8 @@ impl GcMaster {
continue;
}

let iter = iter_res.unwrap();
for kv in iter {
let mut iter = iter_res.unwrap();
while let Some(kv) = iter.next().await {
let (user_key, version) = KeyDecoder::decode_key_gc_userkey_version(kv.0);

let (slot_range_left, slot_range_right) = self.topo.myself_owned_slots();
Expand Down Expand Up @@ -248,8 +248,8 @@ impl GcWorker {
// delete all data key of this key and version
let bound_range =
KEY_ENCODER.encode_txnkv_hash_data_key_range(&user_key, version);
let iter = txn.scan_keys(bound_range, u32::MAX).await?;
for k in iter {
let mut iter = txn.scan_keys_stream(bound_range, u32::MAX).await?;
while let Some(k) = iter.next().await {
txn.delete(k).await?;
}
}
Expand All @@ -261,8 +261,8 @@ impl GcWorker {
// delete all data key of this key and version
let bound_range =
KEY_ENCODER.encode_txnkv_list_data_key_range(&user_key, version);
let iter = txn.scan_keys(bound_range, u32::MAX).await?;
for k in iter {
let mut iter = txn.scan_keys_stream(bound_range, u32::MAX).await?;
while let Some(k) = iter.next().await {
txn.delete(k).await?;
}
}
Expand All @@ -281,8 +281,8 @@ impl GcWorker {
// delete all data key of this key and version
let bound_range =
KEY_ENCODER.encode_txnkv_set_data_key_range(&user_key, version);
let iter = txn.scan_keys(bound_range, u32::MAX).await?;
for k in iter {
let mut iter = txn.scan_keys_stream(bound_range, u32::MAX).await?;
while let Some(k) = iter.next().await {
txn.delete(k).await?;
}
}
Expand All @@ -304,16 +304,16 @@ impl GcWorker {
&String::from_utf8_lossy(&task.user_key),
task.version,
);
let iter = txn.scan_keys(bound_range, u32::MAX).await?;
for k in iter {
let mut iter = txn.scan_keys_stream(bound_range, u32::MAX).await?;
while let Some(k) = iter.next().await {
txn.delete(k).await?;
}

// delete all data key of this key and version
let bound_range =
KEY_ENCODER.encode_txnkv_zset_data_key_range(&user_key, version);
let iter = txn.scan_keys(bound_range, u32::MAX).await?;
for k in iter {
let mut iter = txn.scan_keys_stream(bound_range, u32::MAX).await?;
while let Some(k) = iter.next().await {
txn.delete(k).await?;
}
}
Expand Down
3 changes: 3 additions & 0 deletions src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,9 @@ pub mod cluster;

pub mod client;

pub mod rdb;
use rdb::RDB;

pub mod utils;

pub mod config;
Expand Down
Loading