Mirrors/continuwuity
0
0 Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2026-11-14 12:46:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: Merge origin/main into nex/perf/get-missing-events

Browse Source
This commit is contained in:
timedout committed 2026-06-28 01:29:23 +01:00
2 parents 3b6858e936 458e0b53ac
commit d719fe2048
160 files changed
+16120 -15730
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/database/engine/backup.rs
+70 -74
View File
@@ -1,92 +1,88 @@
use std::{ffi::OsString, path::PathBuf};
use conduwuit::{Err, Result, error, implement, info, utils::time::rfc2822_from_seconds, warn};
use conduwuit::{Err, Result, error, info, utils::time::rfc2822_from_seconds, warn};
use rocksdb::backup::{BackupEngine, BackupEngineOptions};
use super::Engine;
use crate::util::map_err;
#[implement(Engine)]
#[tracing::instrument(skip(self), level = "info")]
pub fn backup(&self) -> Result {
let mut engine = self.backup_engine()?;
let config = &self.ctx.server.config;
if config.database_backups_to_keep > 0 {
engine
.create_new_backup_flush(&self.db, true)
.map_err(map_err)?;
impl super::Engine {
#[tracing::instrument(skip(self), level = "info")]
pub fn backup(&self) -> Result {
let mut engine = self.backup_engine()?;
let config = &self.ctx.server.config;
if config.database_backups_to_keep > 0 {
engine
.create_new_backup_flush(&self.db, true)
.map_err(map_err)?;
let engine_info = engine.get_backup_info();
let info = &engine_info.last().expect("backup engine info is not empty");
info!(
"Created database backup #{} using {} bytes in {} files",
info.backup_id, info.size, info.num_files,
);
}
if config.database_backups_to_keep >= 0 {
let keep = u32::try_from(config.database_backups_to_keep)?;
if let Err(e) = engine.purge_old_backups(keep.try_into()?) {
error!("Failed to purge old backup: {e:?}");
let engine_info = engine.get_backup_info();
let info = &engine_info.last().expect("backup engine info is not empty");
info!(
"Created database backup #{} using {} bytes in {} files",
info.backup_id, info.size, info.num_files,
);
}
if config.database_backups_to_keep >= 0 {
let keep = u32::try_from(config.database_backups_to_keep)?;
if let Err(e) = engine.purge_old_backups(keep.try_into()?) {
error!("Failed to purge old backup: {e:?}");
}
}
if config.database_backups_to_keep == 0 {
warn!("Configuration item `database_backups_to_keep` is set to 0.");
}
Ok(())
}
if config.database_backups_to_keep == 0 {
warn!("Configuration item `database_backups_to_keep` is set to 0.");
pub fn backup_list(&self) -> Result<impl Iterator<Item = String> + Send> {
let info = self.backup_engine()?.get_backup_info();
if info.is_empty() {
return Err!("No backups found.");
}
let list = info.into_iter().map(|info| {
format!(
"#{} {}: {} bytes, {} files",
info.backup_id,
rfc2822_from_seconds(info.timestamp),
info.size,
info.num_files,
)
});
Ok(list)
}
Ok(())
}
pub fn backup_count(&self) -> Result<usize> {
let info = self.backup_engine()?.get_backup_info();
#[implement(Engine)]
pub fn backup_list(&self) -> Result<impl Iterator<Item = String> + Send> {
let info = self.backup_engine()?.get_backup_info();
if info.is_empty() {
return Err!("No backups found.");
Ok(info.len())
}
let list = info.into_iter().map(|info| {
format!(
"#{} {}: {} bytes, {} files",
info.backup_id,
rfc2822_from_seconds(info.timestamp),
info.size,
info.num_files,
)
});
Ok(list)
}
#[implement(Engine)]
pub fn backup_count(&self) -> Result<usize> {
let info = self.backup_engine()?.get_backup_info();
Ok(info.len())
}
#[implement(Engine)]
fn backup_engine(&self) -> Result<BackupEngine> {
let path = self.backup_path()?;
let options = BackupEngineOptions::new(path).map_err(map_err)?;
BackupEngine::open(&options, &self.ctx.env.lock()).map_err(map_err)
}
#[implement(Engine)]
fn backup_path(&self) -> Result<OsString> {
let path = self
.ctx
.server
.config
.database_backup_path
.clone()
.map(PathBuf::into_os_string)
.unwrap_or_default();
if path.is_empty() {
return Err!(Config("database_backup_path", "Configure path to enable backups"));
fn backup_engine(&self) -> Result<BackupEngine> {
let path = self.backup_path()?;
let options = BackupEngineOptions::new(path).map_err(map_err)?;
BackupEngine::open(&options, &self.ctx.env.lock()).map_err(map_err)
}
Ok(path)
fn backup_path(&self) -> Result<OsString> {
let path = self
.ctx
.server
.config
.database_backup_path
.clone()
.map(PathBuf::into_os_string)
.unwrap_or_default();
if path.is_empty() {
return Err!(Config("database_backup_path", "Configure path to enable backups"));
}
Ok(path)
}
}
src/database/engine/files.rs
+10 -10
View File
@@ -1,15 +1,15 @@
use conduwuit::{Result, implement};
use conduwuit::Result;
use rocksdb::LiveFile as SstFile;
use super::Engine;
use crate::util::map_err;
#[implement(Engine)]
pub fn file_list(&self) -> impl Iterator<Item = Result<SstFile>> + Send + use<> {
self.db
.live_files()
.map_err(map_err)
.into_iter()
.flat_map(Vec::into_iter)
.map(Ok)
impl super::Engine {
pub fn file_list(&self) -> impl Iterator<Item = Result<SstFile>> + Send + use<> {
self.db
.live_files()
.map_err(map_err)
.into_iter()
.flat_map(Vec::into_iter)
.map(Ok)
}
}
src/database/engine/memory_usage.rs
+22 -21
View File
@@ -1,30 +1,31 @@
use std::fmt::Write;
use conduwuit::{Result, implement};
use conduwuit::Result;
use rocksdb::perf::get_memory_usage_stats;
use super::Engine;
use crate::or_else;
#[implement(Engine)]
pub fn memory_usage(&self) -> Result<String> {
let mut res = String::new();
let stats = get_memory_usage_stats(Some(&[&self.db]), Some(&[&*self.ctx.row_cache.lock()]))
.or_else(or_else)?;
let mibs = |input| f64::from(u32::try_from(input / 1024).unwrap_or(0)) / 1024.0;
writeln!(
res,
"Memory buffers: {:.2} MiB\nPending write: {:.2} MiB\nTable readers: {:.2} MiB\nRow \
cache: {:.2} MiB",
mibs(stats.mem_table_total),
mibs(stats.mem_table_unflushed),
mibs(stats.mem_table_readers_total),
mibs(u64::try_from(self.ctx.row_cache.lock().get_usage())?),
)?;
impl super::Engine {
pub fn memory_usage(&self) -> Result<String> {
let mut res = String::new();
let stats =
get_memory_usage_stats(Some(&[&self.db]), Some(&[&*self.ctx.row_cache.lock()]))
.or_else(or_else)?;
let mibs = |input| f64::from(u32::try_from(input / 1024).unwrap_or(0)) / 1024.0;
writeln!(
res,
"Memory buffers: {:.2} MiB\nPending write: {:.2} MiB\nTable readers: {:.2} MiB\nRow \
cache: {:.2} MiB",
mibs(stats.mem_table_total),
mibs(stats.mem_table_unflushed),
mibs(stats.mem_table_readers_total),
mibs(u64::try_from(self.ctx.row_cache.lock().get_usage())?),
)?;
for (name, cache) in &*self.ctx.col_cache.lock() {
writeln!(res, "{name} cache: {:.2} MiB", mibs(u64::try_from(cache.get_usage())?))?;
for (name, cache) in &*self.ctx.col_cache.lock() {
writeln!(res, "{name} cache: {:.2} MiB", mibs(u64::try_from(cache.get_usage())?))?;
}
Ok(res)
}
Ok(res)
}
src/database/engine/open.rs
+94 -95
View File
@@ -4,11 +4,11 @@ use std::{
sync::{Arc, atomic::AtomicU32},
};
use conduwuit::{Result, debug, implement, info, warn};
use conduwuit::{Result, debug, info, warn};
use rocksdb::{ColumnFamilyDescriptor, Options};
use super::{
Db, Engine,
Db,
cf_opts::cf_options,
db_opts::db_options,
descriptor::{self, Descriptor},
@@ -16,104 +16,103 @@ use super::{
};
use crate::{Context, or_else};
#[implement(Engine)]
#[tracing::instrument(skip_all, level = "info")]
pub(crate) async fn open(ctx: Arc<Context>, desc: &[Descriptor]) -> Result<Arc<Self>> {
let server = &ctx.server;
let config = &server.config;
let path = &config.database_path;
impl super::Engine {
#[tracing::instrument(skip_all, level = "info")]
pub(crate) async fn open(ctx: Arc<Context>, desc: &[Descriptor]) -> Result<Arc<Self>> {
let server = &ctx.server;
let config = &server.config;
let path = &config.database_path;
let db_opts = db_options(config, &ctx.env.lock(), &ctx.row_cache.lock())?;
let db_opts = db_options(config, &ctx.env.lock(), &ctx.row_cache.lock())?;
let cfds = Self::configure_cfds(&ctx, &db_opts, desc)?;
let num_cfds = cfds.len();
debug!("Configured {num_cfds} column descriptors...");
let cfds = Self::configure_cfds(&ctx, &db_opts, desc)?;
let num_cfds = cfds.len();
debug!("Configured {num_cfds} column descriptors...");
let load_time = std::time::Instant::now();
if config.rocksdb_repair {
repair(&db_opts, &config.database_path)?;
let load_time = std::time::Instant::now();
if config.rocksdb_repair {
repair(&db_opts, &config.database_path)?;
}
debug!("Opening database...");
let db = Db::open_cf_descriptors(&db_opts, path, cfds).or_else(or_else)?;
info!(
columns = num_cfds,
sequence = %db.latest_sequence_number(),
time = ?load_time.elapsed(),
"Opened database."
);
Ok(Arc::new(Self {
db,
pool: ctx.pool.clone(),
ctx: ctx.clone(),
checksums: config.rocksdb_checksums,
corks: AtomicU32::new(0),
}))
}
debug!("Opening database...");
let db = Db::open_cf_descriptors(&db_opts, path, cfds).or_else(or_else)?;
#[tracing::instrument(name = "configure", skip_all, level = "debug")]
fn configure_cfds(
ctx: &Arc<Context>,
db_opts: &Options,
desc: &[Descriptor],
) -> Result<Vec<ColumnFamilyDescriptor>> {
let server = &ctx.server;
let config = &server.config;
let path = &config.database_path;
let existing = Self::discover_cfs(path, db_opts);
info!(
columns = num_cfds,
sequence = %db.latest_sequence_number(),
time = ?load_time.elapsed(),
"Opened database."
);
let creating = desc.iter().filter(|desc| !existing.contains(desc.name));
Ok(Arc::new(Self {
db,
pool: ctx.pool.clone(),
ctx: ctx.clone(),
checksums: config.rocksdb_checksums,
corks: AtomicU32::new(0),
}))
}
#[implement(Engine)]
#[tracing::instrument(name = "configure", skip_all, level = "debug")]
fn configure_cfds(
ctx: &Arc<Context>,
db_opts: &Options,
desc: &[Descriptor],
) -> Result<Vec<ColumnFamilyDescriptor>> {
let server = &ctx.server;
let config = &server.config;
let path = &config.database_path;
let existing = Self::discover_cfs(path, db_opts);
let creating = desc.iter().filter(|desc| !existing.contains(desc.name));
let missing = existing
.iter()
.filter(|&name| name != "default")
.filter(|&name| !desc.iter().any(|desc| desc.name == name));
debug!(
existing = existing.len(),
described = desc.len(),
missing = missing.clone().count(),
creating = creating.clone().count(),
"Discovered database columns"
);
missing.clone().for_each(|name| {
debug!("Found unrecognized column {name:?} in existing database.");
});
creating.map(|desc| desc.name).for_each(|name| {
debug!("Creating new column {name:?} not previously found in existing database.");
});
let missing_descriptors = missing.clone().map(|_| descriptor::DROPPED);
let cfopts: Vec<_> = desc
.iter()
.copied()
.chain(missing_descriptors)
.map(|ref desc| cf_options(ctx, db_opts.clone(), desc))
.collect::<Result<_>>()?;
let cfds: Vec<_> = desc
.iter()
.map(|desc| desc.name)
.map(ToOwned::to_owned)
.chain(missing.cloned())
.zip(cfopts.into_iter())
.map(|(name, opts)| ColumnFamilyDescriptor::new(name, opts))
.collect();
Ok(cfds)
}
#[implement(Engine)]
#[tracing::instrument(name = "discover", skip_all, level = "debug")]
fn discover_cfs(path: &Path, opts: &Options) -> BTreeSet<String> {
Db::list_cf(opts, path)
.unwrap_or_default()
.into_iter()
.collect::<BTreeSet<_>>()
let missing = existing
.iter()
.filter(|&name| name != "default")
.filter(|&name| !desc.iter().any(|desc| desc.name == name));
debug!(
existing = existing.len(),
described = desc.len(),
missing = missing.clone().count(),
creating = creating.clone().count(),
"Discovered database columns"
);
missing.clone().for_each(|name| {
debug!("Found unrecognized column {name:?} in existing database.");
});
creating.map(|desc| desc.name).for_each(|name| {
debug!("Creating new column {name:?} not previously found in existing database.");
});
let missing_descriptors = missing.clone().map(|_| descriptor::DROPPED);
let cfopts: Vec<_> = desc
.iter()
.copied()
.chain(missing_descriptors)
.map(|ref desc| cf_options(ctx, db_opts.clone(), desc))
.collect::<Result<_>>()?;
let cfds: Vec<_> = desc
.iter()
.map(|desc| desc.name)
.map(ToOwned::to_owned)
.chain(missing.cloned())
.zip(cfopts.into_iter())
.map(|(name, opts)| ColumnFamilyDescriptor::new(name, opts))
.collect();
Ok(cfds)
}
#[tracing::instrument(name = "discover", skip_all, level = "debug")]
fn discover_cfs(path: &Path, opts: &Options) -> BTreeSet<String> {
Db::list_cf(opts, path)
.unwrap_or_default()
.into_iter()
.collect::<BTreeSet<_>>()
}
}
src/database/map/clear.rs
+20 -19
View File
@@ -1,30 +1,31 @@
use std::sync::Arc;
use conduwuit::{
Result, implement,
Result,
utils::stream::{ReadyExt, TryIgnore},
};
use futures::{Stream, TryStreamExt};
use crate::keyval::Key;
/// Delete all data stored in this map. !!! USE WITH CAUTION !!!
///
/// See for_clear() with additional details.
#[implement(super::Map)]
#[tracing::instrument(level = "trace")]
pub async fn clear(self: &Arc<Self>) {
self.for_clear().ignore_err().ready_for_each(|_| ()).await;
}
impl super::Map {
/// Delete all data stored in this map. !!! USE WITH CAUTION !!!
///
/// See for_clear() with additional details.
#[tracing::instrument(level = "trace")]
pub async fn clear(self: &Arc<Self>) {
// drops your entire database cutely :3
self.for_clear().ignore_err().ready_for_each(|_| ()).await;
}
/// Delete all data stored in this map. !!! USE WITH CAUTION !!!
///
/// Provides stream of keys undergoing deletion along with any errors.
///
/// Note this operation applies to a snapshot of the data when invoked.
/// Additional data written during or after this call may be missed.
#[implement(super::Map)]
#[tracing::instrument(level = "trace")]
pub fn for_clear(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
self.raw_keys().inspect_ok(|key| self.remove(key))
/// Delete all data stored in this map. !!! USE WITH CAUTION !!!
///
/// Provides stream of keys undergoing deletion along with any errors.
///
/// Note this operation applies to a snapshot of the data when invoked.
/// Additional data written during or after this call may be missed.
#[tracing::instrument(level = "trace")]
pub fn for_clear(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
self.raw_keys().inspect_ok(|key| self.remove(key))
}
}
src/database/map/compact.rs
+36 -35
View File
@@ -1,4 +1,4 @@
use conduwuit::{Err, Result, implement};
use conduwuit::{Err, Result};
use rocksdb::{BottommostLevelCompaction, CompactOptions};
use crate::keyval::KeyBuf;
@@ -23,40 +23,41 @@ pub struct Options {
pub exclusive: bool,
}
#[implement(super::Map)]
#[tracing::instrument(
name = "compact",
level = "info",
skip(self),
fields(%self),
)]
pub fn compact_blocking(&self, opts: Options) -> Result {
let mut co = CompactOptions::default();
co.set_exclusive_manual_compaction(opts.exclusive);
co.set_bottommost_level_compaction(match opts.exhaustive {
| true => BottommostLevelCompaction::Force,
| false => BottommostLevelCompaction::ForceOptimized,
});
impl super::Map {
#[tracing::instrument(
name = "compact",
skip(self),
fields(%self),
)]
pub fn compact_blocking(&self, opts: Options) -> Result {
let mut co = CompactOptions::default();
co.set_exclusive_manual_compaction(opts.exclusive);
co.set_bottommost_level_compaction(match opts.exhaustive {
| true => BottommostLevelCompaction::Force,
| false => BottommostLevelCompaction::ForceOptimized,
});
match opts.level {
| (None, None) => {
co.set_change_level(true);
co.set_target_level(-1);
},
| (None, Some(level)) => {
co.set_change_level(true);
co.set_target_level(level.try_into()?);
},
| (Some(level), None) => {
co.set_change_level(false);
co.set_target_level(level.try_into()?);
},
| (Some(_), Some(_)) => return Err!("compacting between specific levels not supported"),
match opts.level {
| (None, None) => {
co.set_change_level(true);
co.set_target_level(-1);
},
| (None, Some(level)) => {
co.set_change_level(true);
co.set_target_level(level.try_into()?);
},
| (Some(level), None) => {
co.set_change_level(false);
co.set_target_level(level.try_into()?);
},
| (Some(_), Some(_)) =>
return Err!("compacting between specific levels not supported"),
}
self.db
.db
.compact_range_cf_opt(&self.cf(), opts.range.0, opts.range.1, &co);
Ok(())
}
self.db
.db
.compact_range_cf_opt(&self.cf(), opts.range.0, opts.range.1, &co);
Ok(())
}
src/database/map/contains.rs
+82 -85
View File
@@ -3,7 +3,7 @@ use std::{convert::AsRef, fmt::Debug, future::Future, io::Write, sync::Arc};
use conduwuit::{
Result,
arrayvec::ArrayVec,
err, implement,
err,
utils::{future::TryExtExt, result::FlatOk},
};
use futures::FutureExt;
@@ -11,93 +11,90 @@ use serde::Serialize;
use crate::{keyval::KeyBuf, ser};
/// Returns true if the map contains the key.
/// - key is serialized into allocated buffer
/// - harder errors may not be reported
#[inline]
#[implement(super::Map)]
pub fn contains<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = bool> + Send + '_ + use<'_, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = KeyBuf::new();
self.bcontains(key, &mut buf)
}
impl super::Map {
/// Returns true if the map contains the key.
/// - key is serialized into allocated buffer
/// - harder errors may not be reported
#[inline]
pub fn contains<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = bool> + Send + '_ + use<'_, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = KeyBuf::new();
self.bcontains(key, &mut buf)
}
/// Returns true if the map contains the key.
/// - key is serialized into stack-buffer
/// - harder errors will panic
#[inline]
#[implement(super::Map)]
pub fn acontains<const MAX: usize, K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = bool> + Send + '_ + use<'_, MAX, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bcontains(key, &mut buf)
}
/// Returns true if the map contains the key.
/// - key is serialized into stack-buffer
/// - harder errors will panic
#[inline]
pub fn acontains<const MAX: usize, K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = bool> + Send + '_ + use<'_, MAX, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bcontains(key, &mut buf)
}
/// Returns true if the map contains the key.
/// - key is serialized into provided buffer
/// - harder errors will panic
#[implement(super::Map)]
#[tracing::instrument(skip(self, buf), fields(%self), level = "trace")]
pub fn bcontains<K, B>(
self: &Arc<Self>,
key: &K,
buf: &mut B,
) -> impl Future<Output = bool> + Send + '_ + use<'_, K, B>
where
K: Serialize + ?Sized + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize query key");
self.exists(key).is_ok()
}
/// Returns true if the map contains the key.
/// - key is serialized into provided buffer
/// - harder errors will panic
#[tracing::instrument(skip(self, buf), fields(%self), level = "trace")]
pub fn bcontains<K, B>(
self: &Arc<Self>,
key: &K,
buf: &mut B,
) -> impl Future<Output = bool> + Send + '_ + use<'_, K, B>
where
K: Serialize + ?Sized + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize query key");
self.exists(key).is_ok()
}
/// Returns Ok if the map contains the key.
/// - key is raw
#[inline]
#[implement(super::Map)]
pub fn exists<'a, K>(
self: &'a Arc<Self>,
key: &K,
) -> impl Future<Output = Result> + Send + 'a + use<'a, K>
where
K: AsRef<[u8]> + ?Sized + Debug + 'a,
{
self.get(key).map(|res| res.map(|_| ()))
}
/// Returns Ok if the map contains the key.
/// - key is raw
#[inline]
pub fn exists<'a, K>(
self: &'a Arc<Self>,
key: &K,
) -> impl Future<Output = Result> + Send + 'a + use<'a, K>
where
K: AsRef<[u8]> + ?Sized + Debug + 'a,
{
self.get(key).map(|res| res.map(|_| ()))
}
/// Returns Ok if the map contains the key; NotFound otherwise. Harder errors
/// may not always be reported properly.
#[implement(super::Map)]
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn exists_blocking<K>(&self, key: &K) -> Result
where
K: AsRef<[u8]> + ?Sized + Debug,
{
self.maybe_exists(key)
.then(|| self.get_blocking(key))
.flat_ok()
.map(|_| ())
.ok_or_else(|| err!(Request(NotFound("Not found in database"))))
}
/// Returns Ok if the map contains the key; NotFound otherwise. Harder
/// errors may not always be reported properly.
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn exists_blocking<K>(&self, key: &K) -> Result
where
K: AsRef<[u8]> + ?Sized + Debug,
{
self.maybe_exists(key)
.then(|| self.get_blocking(key))
.flat_ok()
.map(|_| ())
.ok_or_else(|| err!(Request(NotFound("Not found in database"))))
}
/// Rocksdb limits this to kBlockCacheTier internally so this is not actually a
/// blocking call; in case that changes we set this as well in our read_options.
#[implement(super::Map)]
pub(crate) fn maybe_exists<K>(&self, key: &K) -> bool
where
K: AsRef<[u8]> + ?Sized,
{
self.db
.db
.key_may_exist_cf_opt(&self.cf(), key, &self.cache_read_options)
/// Rocksdb limits this to kBlockCacheTier internally so this is not
/// actually a blocking call; in case that changes we set this as well in
/// our read_options.
pub(crate) fn maybe_exists<K>(&self, key: &K) -> bool
where
K: AsRef<[u8]> + ?Sized,
{
self.db
.db
.key_may_exist_cf_opt(&self.cf(), key, &self.cache_read_options)
}
}
src/database/map/count.rs
+58 -62
View File
@@ -1,72 +1,68 @@
use std::{fmt::Debug, future::Future, sync::Arc};
use conduwuit::implement;
use futures::stream::StreamExt;
use serde::Serialize;
/// Count the total number of entries in the map.
#[implement(super::Map)]
#[inline]
pub fn count(self: &Arc<Self>) -> impl Future<Output = usize> + Send + '_ {
self.raw_keys().count()
}
impl super::Map {
/// Count the total number of entries in the map.
#[inline]
pub fn count(self: &Arc<Self>) -> impl Future<Output = usize> + Send + '_ {
self.raw_keys().count()
}
/// Count the number of entries in the map starting from a lower-bound.
///
/// - From is a structured key
#[implement(super::Map)]
#[inline]
pub fn count_from<'a, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Future<Output = usize> + Send + 'a + use<'a, P>
where
P: Serialize + ?Sized + Debug + 'a,
{
self.keys_from_raw(from).count()
}
/// Count the number of entries in the map starting from a lower-bound.
///
/// - From is a structured key
#[inline]
pub fn count_from<'a, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Future<Output = usize> + Send + 'a + use<'a, P>
where
P: Serialize + ?Sized + Debug + 'a,
{
self.keys_from_raw(from).count()
}
/// Count the number of entries in the map starting from a lower-bound.
///
/// - From is a raw
#[implement(super::Map)]
#[inline]
pub fn raw_count_from<'a, P>(
self: &'a Arc<Self>,
from: &'a P,
) -> impl Future<Output = usize> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_from(from).count()
}
/// Count the number of entries in the map starting from a lower-bound.
///
/// - From is a raw
#[inline]
pub fn raw_count_from<'a, P>(
self: &'a Arc<Self>,
from: &'a P,
) -> impl Future<Output = usize> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_from(from).count()
}
/// Count the number of entries in the map matching a prefix.
///
/// - Prefix is structured key
#[implement(super::Map)]
#[inline]
pub fn count_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Future<Output = usize> + Send + 'a + use<'a, P>
where
P: Serialize + ?Sized + Debug + 'a,
{
self.keys_prefix_raw(prefix).count()
}
/// Count the number of entries in the map matching a prefix.
///
/// - Prefix is structured key
#[inline]
pub fn count_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Future<Output = usize> + Send + 'a + use<'a, P>
where
P: Serialize + ?Sized + Debug + 'a,
{
self.keys_prefix_raw(prefix).count()
}
/// Count the number of entries in the map matching a prefix.
///
/// - Prefix is raw
#[implement(super::Map)]
#[inline]
pub fn raw_count_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Future<Output = usize> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_prefix(prefix).count()
/// Count the number of entries in the map matching a prefix.
///
/// - Prefix is raw
#[inline]
pub fn raw_count_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Future<Output = usize> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_prefix(prefix).count()
}
}
src/database/map/get.rs
+62 -64
View File
@@ -1,6 +1,6 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Err, Result, err, implement, utils::result::MapExpect};
use conduwuit::{Err, Result, err, utils::result::MapExpect};
use futures::{Future, FutureExt, TryFutureExt, future::ready};
use rocksdb::{DBPinnableSlice, ReadOptions};
use tokio::task;
@@ -10,74 +10,72 @@ use crate::{
util::{is_incomplete, map_err, or_else},
};
/// Fetch a value from the database into cache, returning a reference-handle
/// asynchronously. The key is referenced directly to perform the query.
#[implement(super::Map)]
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn get<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K>
where
K: AsRef<[u8]> + Debug + ?Sized,
{
use crate::pool::Get;
impl super::Map {
/// Fetch a value from the database into cache, returning a reference-handle
/// asynchronously. The key is referenced directly to perform the query.
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn get<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K>
where
K: AsRef<[u8]> + Debug + ?Sized,
{
use crate::pool::Get;
let cached = self.get_cached(key);
if matches!(cached, Err(_) | Ok(Some(_))) {
return task::consume_budget()
.map(move |()| cached.map_expect("data found in cache"))
.boxed();
let cached = self.get_cached(key);
if matches!(cached, Err(_) | Ok(Some(_))) {
return task::consume_budget()
.map(move |()| cached.map_expect("data found in cache"))
.boxed();
}
debug_assert!(matches!(cached, Ok(None)), "expected status Incomplete");
let cmd = Get {
map: self.clone(),
key: [key.as_ref().into()].into(),
res: None,
};
self.db
.pool
.execute_get(cmd)
.and_then(|mut res| ready(res.remove(0)))
.boxed()
}
debug_assert!(matches!(cached, Ok(None)), "expected status Incomplete");
let cmd = Get {
map: self.clone(),
key: [key.as_ref().into()].into(),
res: None,
};
/// Fetch a value from the cache without I/O.
#[tracing::instrument(skip(self, key), name = "cache", level = "trace")]
pub(crate) fn get_cached<K>(&self, key: &K) -> Result<Option<Handle<'_>>>
where
K: AsRef<[u8]> + Debug + ?Sized,
{
let res = self.get_blocking_opts(key, &self.cache_read_options);
cached_handle_from(res)
}
self.db
.pool
.execute_get(cmd)
.and_then(|mut res| ready(res.remove(0)))
.boxed()
}
/// Fetch a value from the database into cache, returning a
/// reference-handle. The key is referenced directly to perform the query.
/// This is a thread- blocking call.
#[tracing::instrument(skip(self, key), name = "blocking", level = "trace")]
pub fn get_blocking<K>(&self, key: &K) -> Result<Handle<'_>>
where
K: AsRef<[u8]> + ?Sized,
{
let res = self.get_blocking_opts(key, &self.read_options);
handle_from(res)
}
/// Fetch a value from the cache without I/O.
#[implement(super::Map)]
#[tracing::instrument(skip(self, key), name = "cache", level = "trace")]
pub(crate) fn get_cached<K>(&self, key: &K) -> Result<Option<Handle<'_>>>
where
K: AsRef<[u8]> + Debug + ?Sized,
{
let res = self.get_blocking_opts(key, &self.cache_read_options);
cached_handle_from(res)
}
/// Fetch a value from the database into cache, returning a reference-handle.
/// The key is referenced directly to perform the query. This is a thread-
/// blocking call.
#[implement(super::Map)]
#[tracing::instrument(skip(self, key), name = "blocking", level = "trace")]
pub fn get_blocking<K>(&self, key: &K) -> Result<Handle<'_>>
where
K: AsRef<[u8]> + ?Sized,
{
let res = self.get_blocking_opts(key, &self.read_options);
handle_from(res)
}
#[implement(super::Map)]
fn get_blocking_opts<K>(
&self,
key: &K,
read_options: &ReadOptions,
) -> Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>
where
K: AsRef<[u8]> + ?Sized,
{
self.db.db.get_pinned_cf_opt(&self.cf(), key, read_options)
fn get_blocking_opts<K>(
&self,
key: &K,
read_options: &ReadOptions,
) -> Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>
where
K: AsRef<[u8]> + ?Sized,
{
self.db.db.get_pinned_cf_opt(&self.cf(), key, read_options)
}
}
#[inline]
src/database/map/get_batch.rs
+51 -52
View File
@@ -1,7 +1,7 @@
use std::{convert::AsRef, sync::Arc};
use conduwuit::{
Result, implement,
Result,
utils::{
IterStream,
stream::{WidebandExt, automatic_amplification, automatic_width},
@@ -34,60 +34,59 @@ where
}
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, keys), level = "trace")]
pub(crate) fn get_batch<'a, S, K>(
self: &'a Arc<Self>,
keys: S,
) -> impl Stream<Item = Result<Handle<'a>>> + Send + 'a
where
S: Stream<Item = K> + Send + 'a,
K: AsRef<[u8]> + Send + Sync + 'a,
{
use crate::pool::Get;
impl super::Map {
#[tracing::instrument(skip(self, keys), level = "trace")]
pub(crate) fn get_batch<'a, S, K>(
self: &'a Arc<Self>,
keys: S,
) -> impl Stream<Item = Result<Handle<'a>>> + Send + 'a
where
S: Stream<Item = K> + Send + 'a,
K: AsRef<[u8]> + Send + Sync + 'a,
{
use crate::pool::Get;
keys.ready_chunks(automatic_amplification())
.widen_then(automatic_width(), |chunk| {
self.db.pool.execute_get(Get {
map: self.clone(),
key: chunk.iter().map(AsRef::as_ref).map(Into::into).collect(),
res: None,
keys.ready_chunks(automatic_amplification())
.widen_then(automatic_width(), |chunk| {
self.db.pool.execute_get(Get {
map: self.clone(),
key: chunk.iter().map(AsRef::as_ref).map(Into::into).collect(),
res: None,
})
})
})
.map_ok(|results| results.into_iter().stream())
.try_flatten()
}
.map_ok(|results| results.into_iter().stream())
.try_flatten()
}
#[implement(super::Map)]
#[tracing::instrument(name = "batch_blocking", level = "trace", skip_all)]
pub(crate) fn get_batch_blocking<'a, I, K>(
&self,
keys: I,
) -> impl Iterator<Item = Result<Handle<'_>>> + Send + use<'_, I, K>
where
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
self.get_batch_blocking_opts(keys, &self.read_options)
.map(handle_from)
}
#[tracing::instrument(name = "batch_blocking", level = "trace", skip_all)]
pub(crate) fn get_batch_blocking<'a, I, K>(
&self,
keys: I,
) -> impl Iterator<Item = Result<Handle<'_>>> + Send + use<'_, I, K>
where
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
self.get_batch_blocking_opts(keys, &self.read_options)
.map(handle_from)
}
#[implement(super::Map)]
fn get_batch_blocking_opts<'a, I, K>(
&self,
keys: I,
read_options: &ReadOptions,
) -> impl Iterator<Item = Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>> + Send + use<'_, I, K>
where
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
// Optimization can be `true` if key vector is pre-sorted **by the column
// comparator**.
const SORTED: bool = false;
fn get_batch_blocking_opts<'a, I, K>(
&self,
keys: I,
read_options: &ReadOptions,
) -> impl Iterator<Item = Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>> + Send + use<'_, I, K>
where
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
// Optimization can be `true` if key vector is pre-sorted **by the column
// comparator**.
const SORTED: bool = false;
self.db
.db
.batched_multi_get_cf_opt(&self.cf(), keys, SORTED, read_options)
.into_iter()
self.db
.db
.batched_multi_get_cf_opt(&self.cf(), keys, SORTED, read_options)
.into_iter()
}
}
src/database/map/insert.rs
+203 -214
View File
@@ -5,7 +5,7 @@
use std::{convert::AsRef, fmt::Debug, io::Write};
use conduwuit::{arrayvec::ArrayVec, implement};
use conduwuit::arrayvec::ArrayVec;
use rocksdb::WriteBatchWithTransaction;
use serde::Serialize;
@@ -15,223 +15,212 @@ use crate::{
util::or_else,
};
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is serialized
#[implement(super::Map)]
#[inline]
pub fn put<K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = KeyBuf::new();
let mut val_buf = ValBuf::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is raw
#[implement(super::Map)]
#[inline]
pub fn put_raw<K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
{
let mut key_buf = KeyBuf::new();
self.bput_raw(key, val, &mut key_buf);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized
#[implement(super::Map)]
#[inline]
pub fn raw_put<K, V>(&self, key: K, val: V)
where
K: AsRef<[u8]>,
V: Serialize,
{
let mut val_buf = ValBuf::new();
self.raw_bput(key, val, &mut val_buf);
}
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is serialized to stack-buffer
#[implement(super::Map)]
#[inline]
pub fn put_aput<const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = KeyBuf::new();
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is serialized
#[implement(super::Map)]
#[inline]
pub fn aput_put<const KMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
let mut val_buf = ValBuf::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is serialized to stack-buffer
#[implement(super::Map)]
#[inline]
pub fn aput<const KMAX: usize, const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is raw
#[implement(super::Map)]
#[inline]
pub fn aput_raw<const KMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
self.bput_raw(key, val, &mut key_buf);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized to stack-buffer
#[implement(super::Map)]
#[inline]
pub fn raw_aput<const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: AsRef<[u8]>,
V: Serialize,
{
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.raw_bput(key, val, &mut val_buf);
}
/// Insert Key/Value
///
/// - Key is serialized to supplied buffer
/// - Val is serialized to supplied buffer
#[implement(super::Map)]
pub fn bput<K, V, Bk, Bv>(&self, key: K, val: V, mut buf: (Bk, Bv))
where
K: Serialize + Debug,
V: Serialize,
Bk: Write + AsRef<[u8]>,
Bv: Write + AsRef<[u8]>,
{
let val = ser::serialize(&mut buf.1, val).expect("failed to serialize insertion val");
self.bput_raw(key, val, &mut buf.0);
}
/// Insert Key/Value
///
/// - Key is serialized to supplied buffer
/// - Val is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self, val, buf), level = "trace")]
pub fn bput_raw<K, V, Bk>(&self, key: K, val: V, mut buf: Bk)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
Bk: Write + AsRef<[u8]>,
{
let key = ser::serialize(&mut buf, key).expect("failed to serialize insertion key");
self.insert(&key, val);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized to supplied buffer
#[implement(super::Map)]
pub fn raw_bput<K, V, Bv>(&self, key: K, val: V, mut buf: Bv)
where
K: AsRef<[u8]>,
V: Serialize,
Bv: Write + AsRef<[u8]>,
{
let val = ser::serialize(&mut buf, val).expect("failed to serialize insertion val");
self.insert(&key, val);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is raw
#[implement(super::Map)]
#[tracing::instrument(skip_all, fields(%self), level = "trace")]
pub fn insert<K, V>(&self, key: &K, val: V)
where
K: AsRef<[u8]> + ?Sized,
V: AsRef<[u8]>,
{
let write_options = &self.write_options;
self.db
.db
.put_cf_opt(&self.cf(), key, val, write_options)
.or_else(or_else)
.expect("database insert error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
impl super::Map {
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is serialized
#[inline]
pub fn put<K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = KeyBuf::new();
let mut val_buf = ValBuf::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
self.watchers.wake(key.as_ref());
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, iter), fields(%self), level = "trace")]
pub fn insert_batch<'a, I, K, V>(&'a self, iter: I)
where
I: Iterator<Item = (K, V)> + Send + Debug,
K: AsRef<[u8]> + Sized + Debug + 'a,
V: AsRef<[u8]> + Sized + 'a,
{
let mut batch = WriteBatchWithTransaction::<false>::default();
for (key, val) in iter {
batch.put_cf(&self.cf(), key.as_ref(), val.as_ref());
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is raw
#[inline]
pub fn put_raw<K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
{
let mut key_buf = KeyBuf::new();
self.bput_raw(key, val, &mut key_buf);
}
let write_options = &self.write_options;
self.db
.db
.write_opt(&batch, write_options)
.or_else(or_else)
.expect("database insert batch error");
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized
#[inline]
pub fn raw_put<K, V>(&self, key: K, val: V)
where
K: AsRef<[u8]>,
V: Serialize,
{
let mut val_buf = ValBuf::new();
self.raw_bput(key, val, &mut val_buf);
}
if !self.db.corked() {
self.db.flush().expect("database flush error");
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is serialized to stack-buffer
#[inline]
pub fn put_aput<const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = KeyBuf::new();
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is serialized
#[inline]
pub fn aput_put<const KMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
let mut val_buf = ValBuf::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is serialized to stack-buffer
#[inline]
pub fn aput<const KMAX: usize, const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is raw
#[inline]
pub fn aput_raw<const KMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
self.bput_raw(key, val, &mut key_buf);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized to stack-buffer
#[inline]
pub fn raw_aput<const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: AsRef<[u8]>,
V: Serialize,
{
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.raw_bput(key, val, &mut val_buf);
}
/// Insert Key/Value
///
/// - Key is serialized to supplied buffer
/// - Val is serialized to supplied buffer
pub fn bput<K, V, Bk, Bv>(&self, key: K, val: V, mut buf: (Bk, Bv))
where
K: Serialize + Debug,
V: Serialize,
Bk: Write + AsRef<[u8]>,
Bv: Write + AsRef<[u8]>,
{
let val = ser::serialize(&mut buf.1, val).expect("failed to serialize insertion val");
self.bput_raw(key, val, &mut buf.0);
}
/// Insert Key/Value
///
/// - Key is serialized to supplied buffer
/// - Val is raw
#[tracing::instrument(skip(self, val, buf), level = "trace")]
pub fn bput_raw<K, V, Bk>(&self, key: K, val: V, mut buf: Bk)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
Bk: Write + AsRef<[u8]>,
{
let key = ser::serialize(&mut buf, key).expect("failed to serialize insertion key");
self.insert(&key, val);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized to supplied buffer
pub fn raw_bput<K, V, Bv>(&self, key: K, val: V, mut buf: Bv)
where
K: AsRef<[u8]>,
V: Serialize,
Bv: Write + AsRef<[u8]>,
{
let val = ser::serialize(&mut buf, val).expect("failed to serialize insertion val");
self.insert(&key, val);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is raw
#[tracing::instrument(skip_all, fields(%self), level = "trace")]
pub fn insert<K, V>(&self, key: &K, val: V)
where
K: AsRef<[u8]> + ?Sized,
V: AsRef<[u8]>,
{
let write_options = &self.write_options;
self.db
.db
.put_cf_opt(&self.cf(), key, val, write_options)
.or_else(or_else)
.expect("database insert error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
}
self.watchers.wake(key.as_ref());
}
#[tracing::instrument(skip(self, iter), fields(%self), level = "trace")]
pub fn insert_batch<'a, I, K, V>(&'a self, iter: I)
where
I: Iterator<Item = (K, V)> + Send + Debug,
K: AsRef<[u8]> + Sized + Debug + 'a,
V: AsRef<[u8]> + Sized + 'a,
{
let mut batch = WriteBatchWithTransaction::<false>::default();
for (key, val) in iter {
batch.put_cf(&self.cf(), key.as_ref(), val.as_ref());
}
let write_options = &self.write_options;
self.db
.db
.write_opt(&batch, write_options)
.or_else(or_else)
.expect("database insert batch error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
}
}
}
src/database/map/keys.rs
+36 -37
View File
@@ -1,6 +1,6 @@
use std::sync::Arc;
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::Deserialize;
@@ -9,43 +9,42 @@ use tokio::task;
use super::stream::is_cached;
use crate::{keyval, keyval::Key, stream};
#[implement(super::Map)]
pub fn keys<'a, K>(self: &'a Arc<Self>) -> impl Stream<Item = Result<Key<'a, K>>> + Send
where
K: Deserialize<'a> + Send,
{
self.raw_keys().map(keyval::result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn raw_keys(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_fwd(None);
return task::consume_budget()
.map(move |()| stream::Keys::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
pub fn keys<'a, K>(self: &'a Arc<Self>) -> impl Stream<Item = Result<Key<'a, K>>> + Send
where
K: Deserialize<'a> + Send,
{
self.raw_keys().map(keyval::result_deserialize_key::<K>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
pub fn raw_keys(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
use crate::pool::Seek;
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Keys<'_>>()
.into_stream()
.try_flatten()
.boxed()
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_fwd(None);
return task::consume_budget()
.map(move |()| stream::Keys::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Keys<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
src/database/map/keys_from.rs
+65 -67
View File
@@ -1,6 +1,6 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::{Deserialize, Serialize};
@@ -11,73 +11,71 @@ use crate::{
stream,
};
#[implement(super::Map)]
pub fn keys_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.keys_from_raw(from).map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn keys_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.raw_keys_from(&key)
}
#[implement(super::Map)]
pub fn keys_raw_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
{
self.raw_keys_from(from).map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn raw_keys_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
return stream::Keys::<'_>::from(state.init_fwd(from.as_ref().into())).boxed();
impl super::Map {
pub fn keys_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.keys_from_raw(from).map(result_deserialize_key::<K>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
#[tracing::instrument(skip(self), level = "trace")]
pub fn keys_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.raw_keys_from(&key)
}
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Keys<'_>>()
.into_stream()
.try_flatten()
.boxed()
pub fn keys_raw_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
{
self.raw_keys_from(from).map(result_deserialize_key::<K>)
}
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn raw_keys_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
return stream::Keys::<'_>::from(state.init_fwd(from.as_ref().into())).boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Keys<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
src/database/map/keys_prefix.rs
+46 -48
View File
@@ -1,59 +1,57 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{Stream, StreamExt, TryStreamExt, future};
use serde::{Deserialize, Serialize};
use crate::keyval::{Key, result_deserialize_key, serialize_key};
#[implement(super::Map)]
pub fn keys_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.keys_prefix_raw(prefix)
.map(result_deserialize_key::<K>)
}
impl super::Map {
pub fn keys_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.keys_prefix_raw(prefix)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn keys_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.raw_keys_from(&key)
.try_take_while(move |k: &Key<'_>| future::ok(k.starts_with(&key)))
}
#[tracing::instrument(skip(self), level = "trace")]
pub fn keys_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.raw_keys_from(&key)
.try_take_while(move |k: &Key<'_>| future::ok(k.starts_with(&key)))
}
#[implement(super::Map)]
pub fn keys_raw_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
{
self.raw_keys_prefix(prefix)
.map(result_deserialize_key::<K>)
}
pub fn keys_raw_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
{
self.raw_keys_prefix(prefix)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
pub fn raw_keys_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_from(prefix)
.try_take_while(|k: &Key<'_>| future::ok(k.starts_with(prefix.as_ref())))
pub fn raw_keys_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_from(prefix)
.try_take_while(|k: &Key<'_>| future::ok(k.starts_with(prefix.as_ref())))
}
}
src/database/map/qry.rs
+46 -46
View File
@@ -1,56 +1,56 @@
use std::{convert::AsRef, fmt::Debug, io::Write, sync::Arc};
use conduwuit::{Result, arrayvec::ArrayVec, implement};
use conduwuit::{Result, arrayvec::ArrayVec};
use futures::Future;
use serde::Serialize;
use crate::{Handle, keyval::KeyBuf, ser};
/// Fetch a value from the database into cache, returning a reference-handle
/// asynchronously. The key is serialized into an allocated buffer to perform
/// the query.
#[implement(super::Map)]
#[inline]
pub fn qry<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = KeyBuf::new();
self.bqry(key, &mut buf)
}
impl super::Map {
/// Fetch a value from the database into cache, returning a reference-handle
/// asynchronously. The key is serialized into an allocated buffer to
/// perform the query.
#[inline]
pub fn qry<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = KeyBuf::new();
self.bqry(key, &mut buf)
}
/// Fetch a value from the database into cache, returning a reference-handle
/// asynchronously. The key is serialized into a fixed-sized buffer to perform
/// the query. The maximum size is supplied as const generic parameter.
#[implement(super::Map)]
#[inline]
pub fn aqry<const MAX: usize, K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, MAX, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bqry(key, &mut buf)
}
/// Fetch a value from the database into cache, returning a reference-handle
/// asynchronously. The key is serialized into a fixed-sized buffer to
/// perform the query. The maximum size is supplied as const generic
/// parameter.
#[inline]
pub fn aqry<const MAX: usize, K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, MAX, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bqry(key, &mut buf)
}
/// Fetch a value from the database into cache, returning a reference-handle
/// asynchronously. The key is serialized into a user-supplied Writer.
#[implement(super::Map)]
#[tracing::instrument(skip(self, buf), level = "trace")]
pub fn bqry<K, B>(
self: &Arc<Self>,
key: &K,
buf: &mut B,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K, B>
where
K: Serialize + ?Sized + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize query key");
self.get(key)
/// Fetch a value from the database into cache, returning a reference-handle
/// asynchronously. The key is serialized into a user-supplied Writer.
#[tracing::instrument(skip(self, buf), level = "trace")]
pub fn bqry<K, B>(
self: &Arc<Self>,
key: &K,
buf: &mut B,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K, B>
where
K: Serialize + ?Sized + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize query key");
self.get(key)
}
}
src/database/map/qry_batch.rs
+26 -25
View File
@@ -1,7 +1,7 @@
use std::{fmt::Debug, sync::Arc};
use conduwuit::{
Result, implement,
Result,
utils::{
IterStream,
stream::{WidebandExt, automatic_amplification, automatic_width},
@@ -32,30 +32,31 @@ where
}
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, keys), level = "trace")]
pub(crate) fn qry_batch<'a, S, K>(
self: &'a Arc<Self>,
keys: S,
) -> impl Stream<Item = Result<Handle<'a>>> + Send + 'a
where
S: Stream<Item = K> + Send + 'a,
K: Serialize + Debug + 'a,
{
use crate::pool::Get;
impl super::Map {
#[tracing::instrument(skip(self, keys), level = "trace")]
pub(crate) fn qry_batch<'a, S, K>(
self: &'a Arc<Self>,
keys: S,
) -> impl Stream<Item = Result<Handle<'a>>> + Send + 'a
where
S: Stream<Item = K> + Send + 'a,
K: Serialize + Debug + 'a,
{
use crate::pool::Get;
keys.ready_chunks(automatic_amplification())
.widen_then(automatic_width(), |chunk| {
let keys = chunk
.iter()
.map(ser::serialize_to::<KeyBuf, _>)
.map(|result| result.expect("failed to serialize query key"))
.collect();
keys.ready_chunks(automatic_amplification())
.widen_then(automatic_width(), |chunk| {
let keys = chunk
.iter()
.map(ser::serialize_to::<KeyBuf, _>)
.map(|result| result.expect("failed to serialize query key"))
.collect();
self.db
.pool
.execute_get(Get { map: self.clone(), key: keys, res: None })
})
.map_ok(|results| results.into_iter().stream())
.try_flatten()
self.db
.pool
.execute_get(Get { map: self.clone(), key: keys, res: None })
})
.map_ok(|results| results.into_iter().stream())
.try_flatten()
}
}
src/database/map/remove.rs
+41 -43
View File
@@ -1,55 +1,53 @@
use std::{convert::AsRef, fmt::Debug, io::Write};
use conduwuit::{arrayvec::ArrayVec, implement};
use conduwuit::arrayvec::ArrayVec;
use serde::Serialize;
use crate::{keyval::KeyBuf, ser, util::or_else};
#[implement(super::Map)]
#[inline]
pub fn del<K>(&self, key: K)
where
K: Serialize + Debug,
{
let mut buf = KeyBuf::new();
self.bdel(key, &mut buf);
}
impl super::Map {
#[inline]
pub fn del<K>(&self, key: K)
where
K: Serialize + Debug,
{
let mut buf = KeyBuf::new();
self.bdel(key, &mut buf);
}
#[implement(super::Map)]
#[inline]
pub fn adel<const MAX: usize, K>(&self, key: K)
where
K: Serialize + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bdel(key, &mut buf);
}
#[inline]
pub fn adel<const MAX: usize, K>(&self, key: K)
where
K: Serialize + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bdel(key, &mut buf);
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, buf), level = "trace")]
pub fn bdel<K, B>(&self, key: K, buf: &mut B)
where
K: Serialize + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize deletion key");
self.remove(key);
}
#[tracing::instrument(skip(self, buf), level = "trace")]
pub fn bdel<K, B>(&self, key: K, buf: &mut B)
where
K: Serialize + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize deletion key");
self.remove(key);
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn remove<K>(&self, key: &K)
where
K: AsRef<[u8]> + ?Sized + Debug,
{
let write_options = &self.write_options;
self.db
.db
.delete_cf_opt(&self.cf(), key, write_options)
.or_else(or_else)
.expect("database remove error");
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn remove<K>(&self, key: &K)
where
K: AsRef<[u8]> + ?Sized + Debug,
{
let write_options = &self.write_options;
self.db
.db
.delete_cf_opt(&self.cf(), key, write_options)
.or_else(or_else)
.expect("database remove error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
}
}
}
src/database/map/rev_keys.rs
+37 -37
View File
@@ -1,6 +1,6 @@
use std::sync::Arc;
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::Deserialize;
@@ -9,43 +9,43 @@ use tokio::task;
use super::rev_stream::is_cached;
use crate::{keyval, keyval::Key, stream};
#[implement(super::Map)]
pub fn rev_keys<'a, K>(self: &'a Arc<Self>) -> impl Stream<Item = Result<Key<'a, K>>> + Send
where
K: Deserialize<'a> + Send,
{
self.rev_raw_keys().map(keyval::result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn rev_raw_keys(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_rev(None);
return task::consume_budget()
.map(move |()| stream::KeysRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
pub fn rev_keys<'a, K>(self: &'a Arc<Self>) -> impl Stream<Item = Result<Key<'a, K>>> + Send
where
K: Deserialize<'a> + Send,
{
self.rev_raw_keys().map(keyval::result_deserialize_key::<K>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn rev_raw_keys(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
use crate::pool::Seek;
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::KeysRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_rev(None);
return task::consume_budget()
.map(move |()| stream::KeysRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::KeysRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
src/database/map/rev_keys_from.rs
+67 -69
View File
@@ -1,6 +1,6 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::{Deserialize, Serialize};
@@ -11,75 +11,73 @@ use crate::{
stream,
};
#[implement(super::Map)]
pub fn rev_keys_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.rev_keys_from_raw(from)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_keys_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.rev_raw_keys_from(&key)
}
#[implement(super::Map)]
pub fn rev_keys_raw_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
{
self.rev_raw_keys_from(from)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn rev_raw_keys_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
return stream::KeysRev::<'_>::from(state.init_rev(from.as_ref().into())).boxed();
impl super::Map {
pub fn rev_keys_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.rev_keys_from_raw(from)
.map(result_deserialize_key::<K>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_keys_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.rev_raw_keys_from(&key)
}
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::KeysRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
pub fn rev_keys_raw_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
{
self.rev_raw_keys_from(from)
.map(result_deserialize_key::<K>)
}
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn rev_raw_keys_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
return stream::KeysRev::<'_>::from(state.init_rev(from.as_ref().into())).boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::KeysRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
src/database/map/rev_keys_prefix.rs
+46 -48
View File
@@ -1,59 +1,57 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{Stream, StreamExt, TryStreamExt, future};
use serde::{Deserialize, Serialize};
use crate::keyval::{Key, result_deserialize_key, serialize_key};
#[implement(super::Map)]
pub fn rev_keys_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.rev_keys_prefix_raw(prefix)
.map(result_deserialize_key::<K>)
}
impl super::Map {
pub fn rev_keys_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.rev_keys_prefix_raw(prefix)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_keys_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.rev_raw_keys_from(&key)
.try_take_while(move |k: &Key<'_>| future::ok(k.starts_with(&key)))
}
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_keys_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.rev_raw_keys_from(&key)
.try_take_while(move |k: &Key<'_>| future::ok(k.starts_with(&key)))
}
#[implement(super::Map)]
pub fn rev_keys_raw_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
{
self.rev_raw_keys_prefix(prefix)
.map(result_deserialize_key::<K>)
}
pub fn rev_keys_raw_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
{
self.rev_raw_keys_prefix(prefix)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
pub fn rev_raw_keys_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.rev_raw_keys_from(prefix)
.try_take_while(|k: &Key<'_>| future::ok(k.starts_with(prefix.as_ref())))
pub fn rev_raw_keys_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.rev_raw_keys_from(prefix)
.try_take_while(|k: &Key<'_>| future::ok(k.starts_with(prefix.as_ref())))
}
}
src/database/map/rev_stream.rs
+47 -47
View File
@@ -1,6 +1,6 @@
use std::sync::Arc;
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::Deserialize;
@@ -8,55 +8,55 @@ use tokio::task;
use crate::{keyval, keyval::KeyVal, stream};
/// Iterate key-value entries in the map from the end.
///
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream<'a, K, V>(
self: &'a Arc<Self>,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send
where
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_raw_stream()
.map(keyval::result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map from the end.
///
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn rev_raw_stream(self: &Arc<Self>) -> impl Stream<Item = Result<KeyVal<'_>>> + Send {
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_rev(None);
return task::consume_budget()
.map(move |()| stream::ItemsRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
/// Iterate key-value entries in the map from the end.
///
/// - Result is deserialized
pub fn rev_stream<'a, K, V>(
self: &'a Arc<Self>,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send
where
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_raw_stream()
.map(keyval::result_deserialize::<K, V>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
/// Iterate key-value entries in the map from the end.
///
/// - Result is raw
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn rev_raw_stream(self: &Arc<Self>) -> impl Stream<Item = Result<KeyVal<'_>>> + Send {
use crate::pool::Seek;
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::ItemsRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_rev(None);
return task::consume_budget()
.map(move |()| stream::ItemsRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::ItemsRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
#[tracing::instrument(
src/database/map/rev_stream_from.rs
+90 -92
View File
@@ -1,6 +1,6 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::{Deserialize, Serialize};
@@ -12,100 +12,98 @@ use crate::{
util::is_incomplete,
};
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is serialized
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_stream_from_raw(from)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is serialized
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_stream_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.rev_raw_stream_from(&key)
}
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is raw
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream_raw_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_raw_stream_from(from)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is raw
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn rev_raw_stream_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
let state = state.init_rev(from.as_ref().into());
return task::consume_budget()
.map(move |()| stream::ItemsRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is serialized
/// - Result is deserialized
pub fn rev_stream_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_stream_from_raw(from)
.map(result_deserialize::<K, V>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is serialized
/// - Result is raw
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_stream_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.rev_raw_stream_from(&key)
}
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::ItemsRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is raw
/// - Result is deserialized
pub fn rev_stream_raw_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_raw_stream_from(from)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is raw
/// - Result is raw
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn rev_raw_stream_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
let state = state.init_rev(from.as_ref().into());
return task::consume_budget()
.map(move |()| stream::ItemsRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::ItemsRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
#[tracing::instrument(
src/database/map/rev_stream_prefix.rs
+64 -66
View File
@@ -1,77 +1,75 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{Stream, StreamExt, TryStreamExt, future};
use serde::{Deserialize, Serialize};
use crate::keyval::{KeyVal, result_deserialize, serialize_key};
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_stream_prefix_raw(prefix)
.map(result_deserialize::<K, V>)
}
impl super::Map {
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is deserialized
pub fn rev_stream_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_stream_prefix_raw(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_stream_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.rev_raw_stream_from(&key)
.try_take_while(move |(k, _): &KeyVal<'_>| future::ok(k.starts_with(&key)))
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is raw
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_stream_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.rev_raw_stream_from(&key)
.try_take_while(move |(k, _): &KeyVal<'_>| future::ok(k.starts_with(&key)))
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream_raw_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
V: Deserialize<'a> + Send + 'a,
{
self.rev_raw_stream_prefix(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is deserialized
pub fn rev_stream_raw_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
V: Deserialize<'a> + Send + 'a,
{
self.rev_raw_stream_prefix(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is raw
#[implement(super::Map)]
pub fn rev_raw_stream_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.rev_raw_stream_from(prefix)
.try_take_while(|(k, _): &KeyVal<'_>| future::ok(k.starts_with(prefix.as_ref())))
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is raw
pub fn rev_raw_stream_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.rev_raw_stream_from(prefix)
.try_take_while(|(k, _): &KeyVal<'_>| future::ok(k.starts_with(prefix.as_ref())))
}
}
src/database/map/stream.rs
+46 -46
View File
@@ -1,6 +1,6 @@
use std::sync::Arc;
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::Deserialize;
@@ -8,54 +8,54 @@ use tokio::task;
use crate::{keyval, keyval::KeyVal, stream};
/// Iterate key-value entries in the map from the beginning.
///
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream<'a, K, V>(
self: &'a Arc<Self>,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send
where
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.raw_stream().map(keyval::result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map from the beginning.
///
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn raw_stream(self: &Arc<Self>) -> impl Stream<Item = Result<KeyVal<'_>>> + Send {
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_fwd(None);
return task::consume_budget()
.map(move |()| stream::Items::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
/// Iterate key-value entries in the map from the beginning.
///
/// - Result is deserialized
pub fn stream<'a, K, V>(
self: &'a Arc<Self>,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send
where
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.raw_stream().map(keyval::result_deserialize::<K, V>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
/// Iterate key-value entries in the map from the beginning.
///
/// - Result is raw
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn raw_stream(self: &Arc<Self>) -> impl Stream<Item = Result<KeyVal<'_>>> + Send {
use crate::pool::Seek;
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Items<'_>>()
.into_stream()
.try_flatten()
.boxed()
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_fwd(None);
return task::consume_budget()
.map(move |()| stream::Items::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Items<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
#[tracing::instrument(
src/database/map/stream_from.rs
+92 -94
View File
@@ -1,6 +1,6 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::{Deserialize, Serialize};
@@ -11,105 +11,103 @@ use crate::{
stream,
};
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is serialized
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.stream_from_raw(from).map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is serialized
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn stream_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.raw_stream_from(&key)
}
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is raw
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream_raw_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.raw_stream_from(from).map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is raw
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn raw_stream_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
let state = state.init_fwd(from.as_ref().into());
return task::consume_budget()
.map(move |()| stream::Items::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is serialized
/// - Result is deserialized
pub fn stream_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.stream_from_raw(from).map(result_deserialize::<K, V>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is serialized
/// - Result is raw
#[tracing::instrument(skip(self), level = "trace")]
pub fn stream_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.raw_stream_from(&key)
}
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Items<'_>>()
.into_stream()
.try_flatten()
.boxed()
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is raw
/// - Result is deserialized
pub fn stream_raw_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.raw_stream_from(from).map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is raw
/// - Result is raw
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn raw_stream_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
let state = state.init_fwd(from.as_ref().into());
return task::consume_budget()
.map(move |()| stream::Items::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Items<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
#[tracing::instrument(
name = "cached",
level = "trace",
skip(map, from),
fields(%map),
name = "cached",
level = "trace",
skip(map, from),
fields(%map),
)]
pub(super) fn is_cached<P>(map: &Arc<super::Map>, from: &P) -> bool
where
src/database/map/stream_prefix.rs
+64 -66
View File
@@ -1,77 +1,75 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{Stream, StreamExt, TryStreamExt, future};
use serde::{Deserialize, Serialize};
use crate::keyval::{KeyVal, result_deserialize, serialize_key};
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.stream_prefix_raw(prefix)
.map(result_deserialize::<K, V>)
}
impl super::Map {
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is deserialized
pub fn stream_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.stream_prefix_raw(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn stream_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.raw_stream_from(&key)
.try_take_while(move |(k, _): &KeyVal<'_>| future::ok(k.starts_with(&key)))
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is raw
#[tracing::instrument(skip(self), level = "trace")]
pub fn stream_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.raw_stream_from(&key)
.try_take_while(move |(k, _): &KeyVal<'_>| future::ok(k.starts_with(&key)))
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream_raw_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
V: Deserialize<'a> + Send + 'a,
{
self.raw_stream_prefix(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is deserialized
pub fn stream_raw_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
V: Deserialize<'a> + Send + 'a,
{
self.raw_stream_prefix(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is raw
#[implement(super::Map)]
pub fn raw_stream_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_stream_from(prefix)
.try_take_while(|(k, _): &KeyVal<'_>| future::ok(k.starts_with(prefix.as_ref())))
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is raw
pub fn raw_stream_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_stream_from(prefix)
.try_take_while(|(k, _): &KeyVal<'_>| future::ok(k.starts_with(prefix.as_ref())))
}
}
src/database/pool.rs
+322 -332
View File
@@ -12,7 +12,7 @@ use std::{
use async_channel::{QueueStrategy, Receiver, RecvError, Sender};
use conduwuit::{
Error, Result, Server, SyncMutex, debug, err, error, implement,
Error, Result, Server, SyncMutex, debug, err, error,
result::DebugInspect,
smallvec::SmallVec,
trace,
@@ -72,29 +72,332 @@ const BATCH_INLINE: usize = 1;
const WORKER_STACK_SIZE: usize = 1_048_576;
const WORKER_NAME: &str = "conduwuit:db";
#[implement(Pool)]
pub(crate) fn new(server: &Arc<Server>) -> Result<Arc<Self>> {
const CHAN_SCHED: (QueueStrategy, QueueStrategy) = (QueueStrategy::Fifo, QueueStrategy::Lifo);
impl Pool {
pub(crate) fn new(server: &Arc<Server>) -> Result<Arc<Self>> {
const CHAN_SCHED: (QueueStrategy, QueueStrategy) =
(QueueStrategy::Fifo, QueueStrategy::Lifo);
let (total_workers, queue_sizes, topology) = configure(server);
let (total_workers, queue_sizes, topology) = configure(server);
let (senders, receivers): (Vec<_>, Vec<_>) = queue_sizes
.into_iter()
.map(|cap| async_channel::bounded_with_queue_strategy(cap, CHAN_SCHED))
.unzip();
let (senders, receivers): (Vec<_>, Vec<_>) = queue_sizes
.into_iter()
.map(|cap| async_channel::bounded_with_queue_strategy(cap, CHAN_SCHED))
.unzip();
let pool = Arc::new(Self {
server: server.clone(),
queues: senders,
workers: Vec::new().into(),
topology,
busy: AtomicUsize::default(),
queued_max: AtomicUsize::default(),
});
let pool = Arc::new(Self {
server: server.clone(),
queues: senders,
workers: Vec::new().into(),
topology,
busy: AtomicUsize::default(),
queued_max: AtomicUsize::default(),
});
pool.spawn_until(&receivers, total_workers)?;
pool.spawn_until(&receivers, total_workers)?;
Ok(pool)
Ok(pool)
}
#[tracing::instrument(skip_all, level = "debug")]
pub(crate) fn close(&self) {
let workers = take(&mut *self.workers.lock());
let senders = self.queues.iter().map(Sender::sender_count).sum::<usize>();
let receivers = self
.queues
.iter()
.map(Sender::receiver_count)
.sum::<usize>();
for queue in &self.queues {
queue.close();
}
if workers.is_empty() {
return;
}
debug!(
senders,
receivers,
queues = self.queues.len(),
workers = workers.len(),
"Closing pool. Waiting for workers to join..."
);
workers
.into_iter()
.map(JoinHandle::join)
.map(|result| result.map_err(Error::from_panic))
.enumerate()
.for_each(|(id, result)| match result {
| Ok(()) => trace!(%id, "worker joined"),
| Err(error) => error!(%id, "worker joined with error: {error}"),
});
}
fn spawn_until(self: &Arc<Self>, recv: &[Receiver<Cmd>], count: usize) -> Result {
let mut workers = self.workers.lock();
while workers.len() < count {
self.clone().spawn_one(&mut workers, recv)?;
}
Ok(())
}
#[tracing::instrument(
name = "spawn",
level = "trace",
skip_all,
fields(id = %workers.len())
)]
fn spawn_one(
self: Arc<Self>,
workers: &mut Vec<JoinHandle<()>>,
recv: &[Receiver<Cmd>],
) -> Result {
debug_assert!(!self.queues.is_empty(), "Must have at least one queue");
debug_assert!(!recv.is_empty(), "Must have at least one receiver");
let id = workers.len();
let group = id.overflowing_rem(self.queues.len()).0;
let recv = recv[group].clone();
let handle = thread::Builder::new()
.name(WORKER_NAME.into())
.stack_size(WORKER_STACK_SIZE)
.spawn(move || self.worker(id, &recv))?;
workers.push(handle);
Ok(())
}
#[tracing::instrument(level = "trace", name = "get", skip(self, cmd))]
pub(crate) async fn execute_get(self: &Arc<Self>, mut cmd: Get) -> Result<BatchResult<'_>> {
let (send, recv) = oneshot::channel();
_ = cmd.res.insert(send);
let queue = self.select_queue();
self.execute(queue, Cmd::Get(cmd))
.and_then(move |()| {
recv.map_ok(into_recv_get)
.map_err(|e| err!(error!("recv failed {e:?}")))
})
.await
}
#[tracing::instrument(level = "trace", name = "iter", skip(self, cmd))]
pub(crate) async fn execute_iter(
self: &Arc<Self>,
mut cmd: Seek,
) -> Result<stream::State<'_>> {
let (send, recv) = oneshot::channel();
_ = cmd.res.insert(send);
let queue = self.select_queue();
self.execute(queue, Cmd::Iter(cmd))
.and_then(|()| {
recv.map_ok(into_recv_seek)
.map_err(|e| err!(error!("recv failed {e:?}")))
})
.await
}
fn select_queue(&self) -> &Sender<Cmd> {
let core_id = get_affinity().next().unwrap_or(0);
let chan_id = self.topology[core_id];
self.queues.get(chan_id).unwrap_or_else(|| &self.queues[0])
}
#[tracing::instrument(
level = "trace",
name = "execute",
skip(self, cmd),
fields(
task = ?tokio::task::try_id(),
receivers = queue.receiver_count(),
queued = queue.len(),
queued_max = self.queued_max.load(Ordering::Relaxed),
),
)]
async fn execute(&self, queue: &Sender<Cmd>, cmd: Cmd) -> Result {
if cfg!(debug_assertions) {
self.queued_max.fetch_max(queue.len(), Ordering::Relaxed);
}
queue
.send(cmd)
.await
.map_err(|e| err!(error!("send failed {e:?}")))
}
#[tracing::instrument(
parent = None,
level = "debug",
skip(self, recv),
fields(
tid = ?thread::current().id(),
),
)]
fn worker(self: Arc<Self>, id: usize, recv: &Receiver<Cmd>) {
self.worker_init(id);
self.worker_loop(recv);
}
fn worker_init(&self, id: usize) {
let group = id.overflowing_rem(self.queues.len()).0;
let affinity = self
.topology
.iter()
.enumerate()
.filter(|_| self.queues.len() > 1)
.filter(|_| self.server.config.db_pool_affinity)
.filter_map(|(core_id, &queue_id)| (group == queue_id).then_some(core_id))
.filter_map(nth_core_available);
// affinity is empty (no-op) if there's only one queue
set_affinity(affinity.clone());
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
if affinity.clone().count() == 1 && conduwuit::alloc::je::is_affine_arena() {
use conduwuit::{
alloc::je::this_thread::{arena_id, set_arena},
result::LogDebugErr,
};
let id = affinity.clone().next().expect("at least one id");
if let Ok(arena) = arena_id() {
if arena != id {
set_arena(id).log_debug_err().ok();
}
}
}
debug!(
%group,
affinity = ?affinity.collect::<Vec<_>>(),
"worker ready"
);
}
fn worker_loop(self: &Arc<Self>, recv: &Receiver<Cmd>) {
// initial +1 needed prior to entering wait
self.busy.fetch_add(1, Ordering::Relaxed);
while let Ok(cmd) = self.worker_wait(recv) {
Self::worker_handle(cmd);
}
}
#[tracing::instrument(
name = "wait",
level = "trace",
skip_all,
fields(
receivers = recv.receiver_count(),
queued = recv.len(),
busy = self.busy.fetch_sub(1, Ordering::Relaxed) - 1,
),
)]
fn worker_wait(self: &Arc<Self>, recv: &Receiver<Cmd>) -> Result<Cmd, RecvError> {
recv.recv_blocking().debug_inspect(|_| {
self.busy.fetch_add(1, Ordering::Relaxed);
})
}
fn worker_handle(cmd: Cmd) {
match cmd {
| Cmd::Get(cmd) if cmd.key.len() == 1 => Self::handle_get(cmd),
| Cmd::Get(cmd) => Self::handle_batch(cmd),
| Cmd::Iter(cmd) => Self::handle_iter(cmd),
}
}
#[tracing::instrument(
name = "iter",
level = "trace",
skip_all,
fields(%cmd.map),
)]
fn handle_iter(mut cmd: Seek) {
let chan = cmd.res.take().expect("missing result channel");
if chan.is_canceled() {
return;
}
let from = cmd.key.as_deref();
let result = match cmd.dir {
| Direction::Forward => cmd.state.init_fwd(from),
| Direction::Reverse => cmd.state.init_rev(from),
};
let chan_result = chan.send(into_send_seek(result));
let _chan_sent = chan_result.is_ok();
}
#[tracing::instrument(
name = "batch",
level = "trace",
skip_all,
fields(
%cmd.map,
keys = %cmd.key.len(),
),
)]
fn handle_batch(mut cmd: Get) {
debug_assert!(cmd.key.len() > 1, "should have more than one key");
debug_assert!(!cmd.key.iter().any(SmallVec::is_empty), "querying for empty key");
let chan = cmd.res.take().expect("missing result channel");
if chan.is_canceled() {
return;
}
let keys = cmd.key.iter();
let result: SmallVec<_> = cmd.map.get_batch_blocking(keys).collect();
let chan_result = chan.send(into_send_get(result));
let _chan_sent = chan_result.is_ok();
}
#[tracing::instrument(
name = "get",
level = "trace",
skip_all,
fields(%cmd.map),
)]
fn handle_get(mut cmd: Get) {
debug_assert!(!cmd.key[0].is_empty(), "querying for empty key");
// Obtain the result channel.
let chan = cmd.res.take().expect("missing result channel");
// It is worth checking if the future was dropped while the command was queued
// so we can bail without paying for any query.
if chan.is_canceled() {
return;
}
// Perform the actual database query. We reuse our database::Map interface but
// limited to the blocking calls, rather than creating another surface directly
// with rocksdb here.
let result = cmd.map.get_blocking(&cmd.key[0]);
// Send the result back to the submitter.
let chan_result = chan.send(into_send_get([result].into()));
// If the future was dropped during the query this will fail acceptably.
let _chan_sent = chan_result.is_ok();
}
}
impl Drop for Pool {
@@ -112,319 +415,6 @@ impl Drop for Pool {
}
}
#[implement(Pool)]
#[tracing::instrument(skip_all, level = "debug")]
pub(crate) fn close(&self) {
let workers = take(&mut *self.workers.lock());
let senders = self.queues.iter().map(Sender::sender_count).sum::<usize>();
let receivers = self
.queues
.iter()
.map(Sender::receiver_count)
.sum::<usize>();
for queue in &self.queues {
queue.close();
}
if workers.is_empty() {
return;
}
debug!(
senders,
receivers,
queues = self.queues.len(),
workers = workers.len(),
"Closing pool. Waiting for workers to join..."
);
workers
.into_iter()
.map(JoinHandle::join)
.map(|result| result.map_err(Error::from_panic))
.enumerate()
.for_each(|(id, result)| match result {
| Ok(()) => trace!(%id, "worker joined"),
| Err(error) => error!(%id, "worker joined with error: {error}"),
});
}
#[implement(Pool)]
fn spawn_until(self: &Arc<Self>, recv: &[Receiver<Cmd>], count: usize) -> Result {
let mut workers = self.workers.lock();
while workers.len() < count {
self.clone().spawn_one(&mut workers, recv)?;
}
Ok(())
}
#[implement(Pool)]
#[tracing::instrument(
name = "spawn",
level = "trace",
skip_all,
fields(id = %workers.len())
)]
fn spawn_one(
self: Arc<Self>,
workers: &mut Vec<JoinHandle<()>>,
recv: &[Receiver<Cmd>],
) -> Result {
debug_assert!(!self.queues.is_empty(), "Must have at least one queue");
debug_assert!(!recv.is_empty(), "Must have at least one receiver");
let id = workers.len();
let group = id.overflowing_rem(self.queues.len()).0;
let recv = recv[group].clone();
let handle = thread::Builder::new()
.name(WORKER_NAME.into())
.stack_size(WORKER_STACK_SIZE)
.spawn(move || self.worker(id, &recv))?;
workers.push(handle);
Ok(())
}
#[implement(Pool)]
#[tracing::instrument(level = "trace", name = "get", skip(self, cmd))]
pub(crate) async fn execute_get(self: &Arc<Self>, mut cmd: Get) -> Result<BatchResult<'_>> {
let (send, recv) = oneshot::channel();
_ = cmd.res.insert(send);
let queue = self.select_queue();
self.execute(queue, Cmd::Get(cmd))
.and_then(move |()| {
recv.map_ok(into_recv_get)
.map_err(|e| err!(error!("recv failed {e:?}")))
})
.await
}
#[implement(Pool)]
#[tracing::instrument(level = "trace", name = "iter", skip(self, cmd))]
pub(crate) async fn execute_iter(self: &Arc<Self>, mut cmd: Seek) -> Result<stream::State<'_>> {
let (send, recv) = oneshot::channel();
_ = cmd.res.insert(send);
let queue = self.select_queue();
self.execute(queue, Cmd::Iter(cmd))
.and_then(|()| {
recv.map_ok(into_recv_seek)
.map_err(|e| err!(error!("recv failed {e:?}")))
})
.await
}
#[implement(Pool)]
fn select_queue(&self) -> &Sender<Cmd> {
let core_id = get_affinity().next().unwrap_or(0);
let chan_id = self.topology[core_id];
self.queues.get(chan_id).unwrap_or_else(|| &self.queues[0])
}
#[implement(Pool)]
#[tracing::instrument(
level = "trace",
name = "execute",
skip(self, cmd),
fields(
task = ?tokio::task::try_id(),
receivers = queue.receiver_count(),
queued = queue.len(),
queued_max = self.queued_max.load(Ordering::Relaxed),
),
)]
async fn execute(&self, queue: &Sender<Cmd>, cmd: Cmd) -> Result {
if cfg!(debug_assertions) {
self.queued_max.fetch_max(queue.len(), Ordering::Relaxed);
}
queue
.send(cmd)
.await
.map_err(|e| err!(error!("send failed {e:?}")))
}
#[implement(Pool)]
#[tracing::instrument(
parent = None,
level = "debug",
skip(self, recv),
fields(
tid = ?thread::current().id(),
),
)]
fn worker(self: Arc<Self>, id: usize, recv: &Receiver<Cmd>) {
self.worker_init(id);
self.worker_loop(recv);
}
#[implement(Pool)]
fn worker_init(&self, id: usize) {
let group = id.overflowing_rem(self.queues.len()).0;
let affinity = self
.topology
.iter()
.enumerate()
.filter(|_| self.queues.len() > 1)
.filter(|_| self.server.config.db_pool_affinity)
.filter_map(|(core_id, &queue_id)| (group == queue_id).then_some(core_id))
.filter_map(nth_core_available);
// affinity is empty (no-op) if there's only one queue
set_affinity(affinity.clone());
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
if affinity.clone().count() == 1 && conduwuit::alloc::je::is_affine_arena() {
use conduwuit::{
alloc::je::this_thread::{arena_id, set_arena},
result::LogDebugErr,
};
let id = affinity.clone().next().expect("at least one id");
if let Ok(arena) = arena_id() {
if arena != id {
set_arena(id).log_debug_err().ok();
}
}
}
debug!(
%group,
affinity = ?affinity.collect::<Vec<_>>(),
"worker ready"
);
}
#[implement(Pool)]
fn worker_loop(self: &Arc<Self>, recv: &Receiver<Cmd>) {
// initial +1 needed prior to entering wait
self.busy.fetch_add(1, Ordering::Relaxed);
while let Ok(cmd) = self.worker_wait(recv) {
Pool::worker_handle(cmd);
}
}
#[implement(Pool)]
#[tracing::instrument(
name = "wait",
level = "trace",
skip_all,
fields(
receivers = recv.receiver_count(),
queued = recv.len(),
busy = self.busy.fetch_sub(1, Ordering::Relaxed) - 1,
),
)]
fn worker_wait(self: &Arc<Self>, recv: &Receiver<Cmd>) -> Result<Cmd, RecvError> {
recv.recv_blocking().debug_inspect(|_| {
self.busy.fetch_add(1, Ordering::Relaxed);
})
}
#[implement(Pool)]
fn worker_handle(cmd: Cmd) {
match cmd {
| Cmd::Get(cmd) if cmd.key.len() == 1 => Pool::handle_get(cmd),
| Cmd::Get(cmd) => Pool::handle_batch(cmd),
| Cmd::Iter(cmd) => Pool::handle_iter(cmd),
}
}
#[implement(Pool)]
#[tracing::instrument(
name = "iter",
level = "trace",
skip_all,
fields(%cmd.map),
)]
fn handle_iter(mut cmd: Seek) {
let chan = cmd.res.take().expect("missing result channel");
if chan.is_canceled() {
return;
}
let from = cmd.key.as_deref();
let result = match cmd.dir {
| Direction::Forward => cmd.state.init_fwd(from),
| Direction::Reverse => cmd.state.init_rev(from),
};
let chan_result = chan.send(into_send_seek(result));
let _chan_sent = chan_result.is_ok();
}
#[implement(Pool)]
#[tracing::instrument(
name = "batch",
level = "trace",
skip_all,
fields(
%cmd.map,
keys = %cmd.key.len(),
),
)]
fn handle_batch(mut cmd: Get) {
debug_assert!(cmd.key.len() > 1, "should have more than one key");
debug_assert!(!cmd.key.iter().any(SmallVec::is_empty), "querying for empty key");
let chan = cmd.res.take().expect("missing result channel");
if chan.is_canceled() {
return;
}
let keys = cmd.key.iter();
let result: SmallVec<_> = cmd.map.get_batch_blocking(keys).collect();
let chan_result = chan.send(into_send_get(result));
let _chan_sent = chan_result.is_ok();
}
#[implement(Pool)]
#[tracing::instrument(
name = "get",
level = "trace",
skip_all,
fields(%cmd.map),
)]
fn handle_get(mut cmd: Get) {
debug_assert!(!cmd.key[0].is_empty(), "querying for empty key");
// Obtain the result channel.
let chan = cmd.res.take().expect("missing result channel");
// It is worth checking if the future was dropped while the command was queued
// so we can bail without paying for any query.
if chan.is_canceled() {
return;
}
// Perform the actual database query. We reuse our database::Map interface but
// limited to the blocking calls, rather than creating another surface directly
// with rocksdb here.
let result = cmd.map.get_blocking(&cmd.key[0]);
// Send the result back to the submitter.
let chan_result = chan.send(into_send_get([result].into()));
// If the future was dropped during the query this will fail acceptably.
let _chan_sent = chan_result.is_ok();
}
fn into_send_get(result: BatchResult<'_>) -> BatchResult<'static> {
// SAFETY: Necessary to send the Handle (rust_rocksdb::PinnableSlice) through
// the channel. The lifetime on the handle is a device by rust-rocksdb to