use crate::env::Env;
use crate::{
DBCommon, Error, RateLimiterMode, ThreadMode,
db::DBInner,
ffi,
ffi_util::{raw_data, to_cpath},
};
use libc::{c_char, c_int, c_uchar, c_void};
use std::ffi::CString;
use std::ops::BitOr;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::path::{Path, PathBuf};
use std::process;
use std::slice;
use std::sync::Arc;
pub struct BackupEngineInfo {
pub timestamp: i64,
pub backup_id: u32,
pub size: u64,
pub num_files: u32,
pub app_metadata: Vec<u8>,
}
pub struct BackupEngine {
inner: *mut ffi::rocksdb_backup_engine_t,
_outlive: Env,
_outlive_backup_env: Option<Env>,
}
pub struct BackupEngineOptions {
inner: *mut ffi::rocksdb_backup_engine_options_t,
backup_env: Option<Env>,
}
pub struct RestoreOptions {
inner: *mut ffi::rocksdb_restore_options_t,
}
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
#[repr(i32)]
pub enum CpuPriority {
KIdle = 0,
KLow = 1,
KNormal = 2,
KHigh = 3,
}
impl From<i32> for CpuPriority {
fn from(value: i32) -> Self {
match value {
0 => Self::KIdle,
1 => Self::KLow,
3 => Self::KHigh,
_ => Self::KNormal,
}
}
}
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
#[repr(i32)]
pub enum RestoreMode {
KKeepLatestDbSessionIdFiles = 1,
KVerifyChecksum = 2,
KPurgeAllFiles = 0xffff,
}
impl From<i32> for RestoreMode {
fn from(value: i32) -> Self {
match value {
1 => Self::KKeepLatestDbSessionIdFiles,
2 => Self::KVerifyChecksum,
_ => Self::KPurgeAllFiles,
}
}
}
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct ShareFilesNaming(c_int);
impl ShareFilesNaming {
pub const LEGACY_CRC32C_AND_FILE_SIZE: Self = Self(1);
pub const USE_DB_SESSION_ID: Self = Self(2);
pub const FLAG_INCLUDE_FILE_SIZE: Self = Self(1 << 31);
pub const MASK_NO_NAMING_FLAGS: Self = Self(0xffff);
pub const MASK_NAMING_FLAGS: Self = Self(!0xffff);
pub const DEFAULT: Self = Self(Self::USE_DB_SESSION_ID.0 | Self::FLAG_INCLUDE_FILE_SIZE.0);
pub const fn bits(self) -> c_int {
self.0
}
pub const fn contains(self, other: Self) -> bool {
(self.0 & other.0) == other.0
}
}
impl BitOr for ShareFilesNaming {
type Output = Self;
fn bitor(self, rhs: Self) -> Self {
Self(self.0 | rhs.0)
}
}
type ProgressFn = dyn Fn() + Send + Sync + 'static;
type ExcludeFilesFn = dyn Fn(&[u8]) -> bool + Send + Sync + 'static;
struct ProgressCallback {
callback: Box<ProgressFn>,
}
struct ExcludeFilesCallback {
callback: Box<ExcludeFilesFn>,
}
unsafe extern "C" fn progress_callback_trampoline(state: *mut c_void) {
let holder = unsafe { &*state.cast::<ProgressCallback>() };
if catch_unwind(AssertUnwindSafe(|| (holder.callback)())).is_err() {
process::abort();
}
}
unsafe extern "C" fn exclude_files_callback_trampoline(
state: *mut c_void,
relative_file: *const c_char,
relative_file_len: usize,
) -> c_uchar {
let holder = unsafe { &*state.cast::<ExcludeFilesCallback>() };
let relative_file = if relative_file_len == 0 {
&[][..]
} else {
unsafe { slice::from_raw_parts(relative_file.cast::<u8>(), relative_file_len) }
};
match catch_unwind(AssertUnwindSafe(|| (holder.callback)(relative_file))) {
Ok(true) => 1,
Ok(false) => 0,
Err(_) => process::abort(),
}
}
pub struct CreateBackupOptions {
inner: *mut ffi::rocksdb_create_backup_options_t,
progress_callback: Option<Arc<ProgressCallback>>,
exclude_files_callback: Option<Arc<ExcludeFilesCallback>>,
}
unsafe impl Send for BackupEngine {}
impl BackupEngine {
pub fn open(opts: &BackupEngineOptions, env: &Env) -> Result<Self, Error> {
let be: *mut ffi::rocksdb_backup_engine_t;
unsafe {
be = ffi_try!(ffi::rocksdb_backup_engine_open_opts(
opts.inner,
env.0.inner
));
}
if be.is_null() {
return Err(Error::new("Could not initialize backup engine.".to_owned()));
}
Ok(Self {
inner: be,
_outlive: env.clone(),
_outlive_backup_env: opts.backup_env.clone(),
})
}
pub fn create_new_backup<T: ThreadMode, D: DBInner>(
&mut self,
db: &DBCommon<T, D>,
) -> Result<(), Error> {
self.create_new_backup_flush(db, false)
}
pub fn create_new_backup_flush<T: ThreadMode, D: DBInner>(
&mut self,
db: &DBCommon<T, D>,
flush_before_backup: bool,
) -> Result<(), Error> {
unsafe {
ffi_try!(ffi::rocksdb_backup_engine_create_new_backup_flush(
self.inner,
db.inner.inner(),
c_uchar::from(flush_before_backup),
));
Ok(())
}
}
pub fn create_new_backup_with_options<T: ThreadMode, D: DBInner>(
&mut self,
db: &DBCommon<T, D>,
opts: &CreateBackupOptions,
) -> Result<u32, Error> {
let mut backup_id: u32 = 0;
unsafe {
ffi_try!(ffi::rocksdb_backup_engine_create_new_backup_with_options(
self.inner,
db.inner.inner(),
opts.inner,
&raw mut backup_id,
));
}
Ok(backup_id)
}
pub fn create_new_backup_with_metadata<T: ThreadMode, D: DBInner>(
&mut self,
db: &DBCommon<T, D>,
opts: &CreateBackupOptions,
app_metadata: &[u8],
) -> Result<u32, Error> {
let (metadata_ptr, metadata_len) = if app_metadata.is_empty() {
(std::ptr::null(), 0)
} else {
(app_metadata.as_ptr().cast::<c_char>(), app_metadata.len())
};
let mut backup_id: u32 = 0;
unsafe {
ffi_try!(ffi::rocksdb_backup_engine_create_new_backup_with_metadata(
self.inner,
db.inner.inner(),
opts.inner,
metadata_ptr,
metadata_len,
&raw mut backup_id,
));
}
Ok(backup_id)
}
pub fn purge_old_backups(&mut self, num_backups_to_keep: usize) -> Result<(), Error> {
unsafe {
ffi_try!(ffi::rocksdb_backup_engine_purge_old_backups(
self.inner,
num_backups_to_keep as u32,
));
Ok(())
}
}
pub fn restore_from_latest_backup<D: AsRef<Path>, W: AsRef<Path>>(
&mut self,
db_dir: D,
wal_dir: W,
opts: &RestoreOptions,
) -> Result<(), Error> {
let c_db_dir = to_cpath(db_dir)?;
let c_wal_dir = to_cpath(wal_dir)?;
unsafe {
ffi_try!(ffi::rocksdb_backup_engine_restore_db_from_latest_backup(
self.inner,
c_db_dir.as_ptr(),
c_wal_dir.as_ptr(),
opts.inner,
));
}
Ok(())
}
pub fn restore_from_backup<D: AsRef<Path>, W: AsRef<Path>>(
&mut self,
db_dir: D,
wal_dir: W,
opts: &RestoreOptions,
backup_id: u32,
) -> Result<(), Error> {
let c_db_dir = to_cpath(db_dir)?;
let c_wal_dir = to_cpath(wal_dir)?;
unsafe {
ffi_try!(ffi::rocksdb_backup_engine_restore_db_from_backup(
self.inner,
c_db_dir.as_ptr(),
c_wal_dir.as_ptr(),
opts.inner,
backup_id,
));
}
Ok(())
}
pub fn verify_backup(&self, backup_id: u32) -> Result<(), Error> {
unsafe {
ffi_try!(ffi::rocksdb_backup_engine_verify_backup(
self.inner, backup_id,
));
}
Ok(())
}
pub fn get_backup_info(&self) -> Vec<BackupEngineInfo> {
unsafe {
let i = ffi::rocksdb_backup_engine_get_backup_info(self.inner);
let n = ffi::rocksdb_backup_engine_info_count(i);
let mut info = Vec::with_capacity(n as usize);
for index in 0..n {
let mut metadata_len: usize = 0;
let metadata_ptr =
ffi::rocksdb_backup_engine_info_app_metadata(i, index, &raw mut metadata_len);
let app_metadata = if metadata_len == 0 {
Vec::new()
} else {
raw_data(metadata_ptr, metadata_len).unwrap_or_default()
};
info.push(BackupEngineInfo {
timestamp: ffi::rocksdb_backup_engine_info_timestamp(i, index),
backup_id: ffi::rocksdb_backup_engine_info_backup_id(i, index),
size: ffi::rocksdb_backup_engine_info_size(i, index),
num_files: ffi::rocksdb_backup_engine_info_number_files(i, index),
app_metadata,
});
}
ffi::rocksdb_backup_engine_info_destroy(i);
info
}
}
pub fn stop_backup(&mut self) {
unsafe {
ffi::rocksdb_backup_engine_stop_backup(self.inner);
}
}
}
impl BackupEngineOptions {
pub fn new<P: AsRef<Path>>(backup_dir: P) -> Result<Self, Error> {
let backup_dir = backup_dir.as_ref();
let c_backup_dir = CString::new(backup_dir.to_string_lossy().as_bytes()).map_err(|_| {
Error::new(
"Failed to convert backup_dir to CString \
when constructing BackupEngineOptions"
.to_owned(),
)
})?;
unsafe {
let opts = ffi::rocksdb_backup_engine_options_create(c_backup_dir.as_ptr());
assert!(!opts.is_null(), "Could not create RocksDB backup options");
Ok(Self {
inner: opts,
backup_env: None,
})
}
}
pub fn set_max_background_operations(&mut self, max_background_operations: i32) {
unsafe {
ffi::rocksdb_backup_engine_options_set_max_background_operations(
self.inner,
max_background_operations,
);
}
}
pub fn set_sync(&mut self, sync: bool) {
unsafe {
ffi::rocksdb_backup_engine_options_set_sync(self.inner, c_uchar::from(sync));
}
}
pub fn get_sync(&mut self) -> bool {
let val_u8 = unsafe { ffi::rocksdb_backup_engine_options_get_sync(self.inner) };
val_u8 != 0
}
pub fn set_current_temperatures_override_manifest(&mut self, val: bool) {
unsafe {
ffi::rocksdb_backup_engine_options_set_current_temperatures_override_manifest(
self.inner,
c_uchar::from(val),
);
}
}
pub fn get_current_temperatures_override_manifest(&self) -> bool {
unsafe {
ffi::rocksdb_backup_engine_options_get_current_temperatures_override_manifest(
self.inner,
) != 0
}
}
pub fn set_io_buffer_size(&mut self, val: u64) {
unsafe {
ffi::rocksdb_backup_engine_options_set_io_buffer_size(self.inner, val);
}
}
pub fn get_io_buffer_size(&self) -> u64 {
unsafe { ffi::rocksdb_backup_engine_options_get_io_buffer_size(self.inner) }
}
pub fn set_schema_version(&mut self, val: c_int) {
unsafe {
ffi::rocksdb_backup_engine_options_set_schema_version(self.inner, val);
}
}
pub fn get_schema_version(&self) -> c_int {
unsafe { ffi::rocksdb_backup_engine_options_get_schema_version(self.inner) }
}
pub fn set_share_files_with_checksum(&mut self, val: bool) {
unsafe {
ffi::rocksdb_backup_engine_options_set_share_files_with_checksum(
self.inner,
c_uchar::from(val),
);
}
}
pub fn get_share_files_with_checksum(&self) -> bool {
unsafe { ffi::rocksdb_backup_engine_options_get_share_files_with_checksum(self.inner) != 0 }
}
pub fn get_backup_log_files(&self) -> bool {
unsafe { ffi::rocksdb_backup_engine_options_get_backup_log_files(self.inner) != 0 }
}
pub fn get_backup_rate_limit(&self) -> u64 {
unsafe { ffi::rocksdb_backup_engine_options_get_backup_rate_limit(self.inner) }
}
pub fn get_callback_trigger_interval_size(&self) -> u64 {
unsafe { ffi::rocksdb_backup_engine_options_get_callback_trigger_interval_size(self.inner) }
}
pub fn get_destroy_old_data(&self) -> bool {
unsafe { ffi::rocksdb_backup_engine_options_get_destroy_old_data(self.inner) != 0 }
}
pub fn get_max_background_operations(&self) -> c_int {
unsafe { ffi::rocksdb_backup_engine_options_get_max_background_operations(self.inner) }
}
pub fn get_max_valid_backups_to_open(&self) -> c_int {
unsafe { ffi::rocksdb_backup_engine_options_get_max_valid_backups_to_open(self.inner) }
}
pub fn get_restore_rate_limit(&self) -> u64 {
unsafe { ffi::rocksdb_backup_engine_options_get_restore_rate_limit(self.inner) }
}
pub fn get_share_table_files(&self) -> bool {
unsafe { ffi::rocksdb_backup_engine_options_get_share_table_files(self.inner) != 0 }
}
pub fn set_backup_log_files(&mut self, val: bool) {
unsafe {
ffi::rocksdb_backup_engine_options_set_backup_log_files(self.inner, c_uchar::from(val));
}
}
pub fn set_backup_rate_limit(&mut self, val: u64) {
unsafe {
ffi::rocksdb_backup_engine_options_set_backup_rate_limit(self.inner, val);
}
}
pub fn set_callback_trigger_interval_size(&mut self, val: u64) {
unsafe {
ffi::rocksdb_backup_engine_options_set_callback_trigger_interval_size(self.inner, val);
}
}
pub fn set_destroy_old_data(&mut self, val: bool) {
unsafe {
ffi::rocksdb_backup_engine_options_set_destroy_old_data(self.inner, c_uchar::from(val));
}
}
pub fn set_max_valid_backups_to_open(&mut self, val: c_int) {
unsafe {
ffi::rocksdb_backup_engine_options_set_max_valid_backups_to_open(self.inner, val);
}
}
pub fn set_restore_rate_limit(&mut self, val: u64) {
unsafe {
ffi::rocksdb_backup_engine_options_set_restore_rate_limit(self.inner, val);
}
}
pub fn set_share_table_files(&mut self, val: bool) {
unsafe {
ffi::rocksdb_backup_engine_options_set_share_table_files(
self.inner,
c_uchar::from(val),
);
}
}
pub fn set_backup_dir<P: AsRef<Path>>(&mut self, backup_dir: P) -> Result<(), Error> {
let c_backup_dir = to_cpath(backup_dir)?;
unsafe {
ffi::rocksdb_backup_engine_options_set_backup_dir(self.inner, c_backup_dir.as_ptr());
}
Ok(())
}
pub fn get_backup_dir(&self) -> PathBuf {
unsafe {
let mut len: usize = 0;
let ptr = ffi::rocksdb_backup_engine_options_get_backup_dir(self.inner, &raw mut len);
if len == 0 {
return PathBuf::new();
}
path_from_bytes(slice::from_raw_parts(ptr.cast::<u8>(), len))
}
}
pub fn set_share_files_with_checksum_naming(&mut self, val: ShareFilesNaming) {
unsafe {
ffi::rocksdb_backup_engine_options_set_share_files_with_checksum_naming(
self.inner,
val.bits(),
);
}
}
pub fn get_share_files_with_checksum_naming(&self) -> ShareFilesNaming {
unsafe {
ShareFilesNaming(
ffi::rocksdb_backup_engine_options_get_share_files_with_checksum_naming(self.inner),
)
}
}
pub fn set_env(&mut self, env: &Env) {
unsafe {
ffi::rocksdb_backup_engine_options_set_env(self.inner, env.0.inner);
}
self.backup_env = Some(env.clone());
}
pub fn set_backup_rate_limiter(
&mut self,
rate_bytes_per_sec: i64,
refill_period_us: i64,
fairness: i32,
) {
unsafe {
let ratelimiter =
ffi::rocksdb_ratelimiter_create(rate_bytes_per_sec, refill_period_us, fairness);
ffi::rocksdb_backup_engine_options_set_backup_rate_limiter(self.inner, ratelimiter);
ffi::rocksdb_ratelimiter_destroy(ratelimiter);
}
}
pub fn set_backup_rate_limiter_with_mode(
&mut self,
rate_bytes_per_sec: i64,
refill_period_us: i64,
fairness: i32,
mode: RateLimiterMode,
auto_tuned: bool,
) {
unsafe {
let ratelimiter = ffi::rocksdb_ratelimiter_create_with_mode(
rate_bytes_per_sec,
refill_period_us,
fairness,
mode as c_int,
auto_tuned,
);
ffi::rocksdb_backup_engine_options_set_backup_rate_limiter(self.inner, ratelimiter);
ffi::rocksdb_ratelimiter_destroy(ratelimiter);
}
}
pub fn set_restore_rate_limiter(
&mut self,
rate_bytes_per_sec: i64,
refill_period_us: i64,
fairness: i32,
) {
unsafe {
let ratelimiter =
ffi::rocksdb_ratelimiter_create(rate_bytes_per_sec, refill_period_us, fairness);
ffi::rocksdb_backup_engine_options_set_restore_rate_limiter(self.inner, ratelimiter);
ffi::rocksdb_ratelimiter_destroy(ratelimiter);
}
}
pub fn set_restore_rate_limiter_with_mode(
&mut self,
rate_bytes_per_sec: i64,
refill_period_us: i64,
fairness: i32,
mode: RateLimiterMode,
auto_tuned: bool,
) {
unsafe {
let ratelimiter = ffi::rocksdb_ratelimiter_create_with_mode(
rate_bytes_per_sec,
refill_period_us,
fairness,
mode as c_int,
auto_tuned,
);
ffi::rocksdb_backup_engine_options_set_restore_rate_limiter(self.inner, ratelimiter);
ffi::rocksdb_ratelimiter_destroy(ratelimiter);
}
}
}
#[cfg(unix)]
fn path_from_bytes(bytes: &[u8]) -> PathBuf {
use std::os::unix::ffi::OsStrExt;
PathBuf::from(std::ffi::OsStr::from_bytes(bytes))
}
#[cfg(not(unix))]
fn path_from_bytes(bytes: &[u8]) -> PathBuf {
PathBuf::from(String::from_utf8_lossy(bytes).into_owned())
}
impl CreateBackupOptions {
pub fn set_flush_before_backup(&mut self, val: bool) {
unsafe {
ffi::rocksdb_create_backup_options_set_flush_before_backup(
self.inner,
c_uchar::from(val),
);
}
}
pub fn get_flush_before_backup(&self) -> bool {
unsafe { ffi::rocksdb_create_backup_options_get_flush_before_backup(self.inner) != 0 }
}
pub fn set_atomic_flush(&mut self, val: bool) {
unsafe {
ffi::rocksdb_create_backup_options_set_atomic_flush(self.inner, c_uchar::from(val));
}
}
pub fn get_atomic_flush(&self) -> bool {
unsafe { ffi::rocksdb_create_backup_options_get_atomic_flush(self.inner) != 0 }
}
pub fn set_decrease_background_thread_cpu_priority(&mut self, val: bool) {
unsafe {
ffi::rocksdb_create_backup_options_set_decrease_background_thread_cpu_priority(
self.inner,
c_uchar::from(val),
);
}
}
pub fn get_decrease_background_thread_cpu_priority(&self) -> bool {
unsafe {
ffi::rocksdb_create_backup_options_get_decrease_background_thread_cpu_priority(
self.inner,
) != 0
}
}
pub fn set_background_thread_cpu_priority(&mut self, val: CpuPriority) {
unsafe {
ffi::rocksdb_create_backup_options_set_background_thread_cpu_priority(
self.inner,
val as c_int,
);
}
}
pub fn get_background_thread_cpu_priority(&self) -> CpuPriority {
unsafe {
CpuPriority::from(
ffi::rocksdb_create_backup_options_get_background_thread_cpu_priority(self.inner),
)
}
}
pub fn set_progress_callback<F>(&mut self, callback: F)
where
F: Fn() + Send + Sync + 'static,
{
let holder = Arc::new(ProgressCallback {
callback: Box::new(callback),
});
let state = std::ptr::from_ref::<ProgressCallback>(holder.as_ref())
.cast::<c_void>()
.cast_mut();
unsafe {
ffi::rocksdb_create_backup_options_set_progress_callback(
self.inner,
state,
Some(progress_callback_trampoline),
);
}
self.progress_callback = Some(holder);
}
pub fn set_exclude_files_callback<F>(&mut self, callback: F)
where
F: Fn(&[u8]) -> bool + Send + Sync + 'static,
{
let holder = Arc::new(ExcludeFilesCallback {
callback: Box::new(callback),
});
let state = std::ptr::from_ref::<ExcludeFilesCallback>(holder.as_ref())
.cast::<c_void>()
.cast_mut();
unsafe {
ffi::rocksdb_create_backup_options_set_exclude_files_callback(
self.inner,
state,
Some(exclude_files_callback_trampoline),
);
}
self.exclude_files_callback = Some(holder);
}
}
impl Default for CreateBackupOptions {
fn default() -> Self {
unsafe {
let opts = ffi::rocksdb_create_backup_options_create();
assert!(
!opts.is_null(),
"Could not create RocksDB create backup options"
);
Self {
inner: opts,
progress_callback: None,
exclude_files_callback: None,
}
}
}
}
impl RestoreOptions {
pub fn set_keep_log_files(&mut self, keep_log_files: bool) {
unsafe {
ffi::rocksdb_restore_options_set_keep_log_files(self.inner, i32::from(keep_log_files));
}
}
pub fn get_keep_log_files(&self) -> bool {
unsafe { ffi::rocksdb_restore_options_get_keep_log_files(self.inner) != 0 }
}
pub fn set_mode(&mut self, mode: RestoreMode) {
unsafe {
ffi::rocksdb_restore_options_set_mode(self.inner, mode as c_int);
}
}
pub fn get_mode(&self) -> RestoreMode {
unsafe { RestoreMode::from(ffi::rocksdb_restore_options_get_mode(self.inner)) }
}
}
impl Default for RestoreOptions {
fn default() -> Self {
unsafe {
let opts = ffi::rocksdb_restore_options_create();
assert!(!opts.is_null(), "Could not create RocksDB restore options");
Self { inner: opts }
}
}
}
impl Drop for BackupEngine {
fn drop(&mut self) {
unsafe {
ffi::rocksdb_backup_engine_close(self.inner);
}
}
}
impl Drop for BackupEngineOptions {
fn drop(&mut self) {
unsafe {
ffi::rocksdb_backup_engine_options_destroy(self.inner);
}
}
}
impl Drop for CreateBackupOptions {
fn drop(&mut self) {
unsafe {
ffi::rocksdb_create_backup_options_destroy(self.inner);
}
}
}
impl Drop for RestoreOptions {
fn drop(&mut self) {
unsafe {
ffi::rocksdb_restore_options_destroy(self.inner);
}
}
}
#[cfg(test)]
mod tests {
use super::BackupEngineOptions;
#[test]
fn test_sync() {
let dir = tempfile::Builder::new()
.prefix("rocksdb-test-sync")
.tempdir()
.expect("Failed to create temporary path for db.");
let mut opts = BackupEngineOptions::new(dir.path()).unwrap();
assert!(opts.get_sync());
opts.set_sync(false);
assert!(!opts.get_sync());
}
}