Compare commits

..

2 Commits

Author SHA1 Message Date
dignifiedquire
c2e275cb96 array backed pool 2023-02-18 16:26:58 +01:00
link2xt
b2c5560e30 Remove try_many_times r2d2 bug workaround 2023-02-18 01:57:19 +00:00
4 changed files with 42 additions and 43 deletions

View File

@@ -61,7 +61,7 @@ pretty_env_logger = { version = "0.4", optional = true }
quick-xml = "0.27"
rand = "0.8"
regex = "1.7"
rusqlite = { version = "0.28", features = ["sqlcipher", "release_memory"] }
rusqlite = { version = "0.28", features = ["sqlcipher"] }
rust-hsluv = "0.1"
sanitize-filename = "0.4"
serde_json = "1.0"

View File

@@ -1,7 +1,6 @@
//! # Account manager module.
use std::collections::BTreeMap;
use std::future::Future;
use std::path::{Path, PathBuf};
use anyhow::{ensure, Context as _, Result};
@@ -151,7 +150,7 @@ impl Accounts {
if let Some(cfg) = self.config.get_account(id) {
let account_path = self.dir.join(cfg.dir);
try_many_times(|| fs::remove_dir_all(&account_path))
fs::remove_dir_all(&account_path)
.await
.context("failed to remove account data")?;
}
@@ -187,10 +186,10 @@ impl Accounts {
fs::create_dir_all(self.dir.join(&account_config.dir))
.await
.context("failed to create dir")?;
try_many_times(|| fs::rename(&dbfile, &new_dbfile))
fs::rename(&dbfile, &new_dbfile)
.await
.context("failed to rename dbfile")?;
try_many_times(|| fs::rename(&blobdir, &new_blobdir))
fs::rename(&blobdir, &new_blobdir)
.await
.context("failed to rename blobdir")?;
if walfile.exists() {
@@ -215,7 +214,7 @@ impl Accounts {
}
Err(err) => {
let account_path = std::path::PathBuf::from(&account_config.dir);
try_many_times(|| fs::remove_dir_all(&account_path))
fs::remove_dir_all(&account_path)
.await
.context("failed to remove account data")?;
self.config.remove_account(account_config.id).await?;
@@ -472,33 +471,6 @@ impl Config {
}
}
/// Spend up to 1 minute trying to do the operation.
///
/// Files may remain locked up to 30 seconds due to r2d2 bug:
/// <https://github.com/sfackler/r2d2/issues/99>
async fn try_many_times<F, Fut, T>(f: F) -> std::result::Result<(), T>
where
F: Fn() -> Fut,
Fut: Future<Output = std::result::Result<(), T>>,
{
let mut counter = 0;
loop {
counter += 1;
if let Err(err) = f().await {
if counter > 60 {
return Err(err);
}
// Wait 1 second and try again.
tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
} else {
break;
}
}
Ok(())
}
/// Configuration of a single account.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
struct AccountConfig {

View File

@@ -196,12 +196,12 @@ impl Sql {
/// Creates a new connection pool.
fn new_pool(dbfile: &Path, passphrase: String) -> Result<Pool> {
let mut connections = Vec::new();
for _ in 0..3 {
let connection = new_connection(dbfile, &passphrase)?;
connections.push(connection);
}
let mut connections = [None, None, None]; // array from iter is not stable yet
for c in &mut connections {
let connection = new_connection(dbfile, &passphrase)?;
*c = Some(connection);
}
let pool = Pool::new(connections);
Ok(pool)
}

View File

@@ -7,10 +7,13 @@ use std::sync::{Arc, Weak};
use parking_lot::{Condvar, Mutex};
use rusqlite::Connection;
/// Total size of the connection pool.
pub const POOL_SIZE: usize = 3;
/// Inner connection pool.
struct InnerPool {
/// Available connections.
connections: Mutex<Vec<Connection>>,
connections: Mutex<[Option<Connection>; POOL_SIZE]>,
/// Conditional variable to notify about added connections.
///
@@ -24,7 +27,19 @@ impl InnerPool {
/// The connection could be new or returned back.
fn put(&self, connection: Connection) {
let mut connections = self.connections.lock();
connections.push(connection);
let mut found_one = false;
for c in &mut *connections {
if c.is_none() {
*c = Some(connection);
found_one = true;
break;
}
}
assert!(
found_one,
"attempted to put more connections than available"
);
drop(connections);
self.cond.notify_one();
}
@@ -44,7 +59,6 @@ impl Drop for PooledConnection {
// Put the connection back unless the pool is already dropped.
if let Some(pool) = self.pool.upgrade() {
if let Some(conn) = self.conn.take() {
conn.release_memory().ok();
pool.put(conn);
}
}
@@ -80,7 +94,7 @@ impl fmt::Debug for Pool {
impl Pool {
/// Creates a new connection pool.
pub fn new(connections: Vec<Connection>) -> Self {
pub fn new(connections: [Option<Connection>; POOL_SIZE]) -> Self {
let inner = Arc::new(InnerPool {
connections: Mutex::new(connections),
cond: Condvar::new(),
@@ -93,7 +107,7 @@ impl Pool {
let mut connections = self.inner.connections.lock();
loop {
if let Some(conn) = connections.pop() {
if let Some(conn) = get_next(&mut *connections) {
return PooledConnection {
pool: Arc::downgrade(&self.inner),
conn: Some(conn),
@@ -104,3 +118,16 @@ impl Pool {
}
}
}
/// Returns the first available connection.
///
/// `None` if no connection is availble.
fn get_next(connections: &mut [Option<Connection>; POOL_SIZE]) -> Option<Connection> {
for c in &mut *connections {
if c.is_some() {
return c.take();
}
}
None
}